goodwe-addon/goodwe_controller/test_price_history.py

"""Runnable check for FCST-04 - the day the fill carries.  `python3 test_price_history.py`

WHAT THIS FILE IS DEFENDING. FCST-03 shipped `fill_from_yesterday()` and the
consumers written for it, and shipped NO CALLER. This file grades the caller:
that a partially-published day-ahead feed now produces a plan covering the whole
requested horizon, that the day it carries survives a restart, and that the
cache cannot decay by being fed its own output.

  AC1  THE HOLE IS CLOSED. `resolve_tariff()` on a feed that has published 10
       of 48 hours returns a series covering 48, the 38 unpublished hours are
       provisional and the 10 published ones are the provider's own objects,
       unchanged and unflagged. Graded against the same call with
       `NoHistory()`, which is the pre-FCST-04 behaviour, so the difference is
       MEASURED rather than described.
  AC2  THE WINDOW STOPPED INCLUDING THE PAST, which is the half of this change
       that moves energy. `load_prices()` returns whatever the feed sent, so at
       09:00 a spread was taken over hours that had already elapsed and the
       "cheapest hour" could be one no cycle can buy in - the trade is then
       silently never taken. The filled series is windowed to [since, until),
       so the firm cheapest hour is a hour still ahead. Graded by finding the
       cheapest firm hour both ways and showing it MOVES.
  AC3  THE CACHE CANNOT DECAY. What is recorded is the provider's read and
       never the filled series; if it were, day 2 would carry day 1's guesses
       and day 3 would carry day 2's carry of them, behind a `provisional` flag
       that stays true while the NUMBER rots. Graded by driving `carry()` twice
       and reading the file back.
  AC4  A STORE THAT CANNOT ANSWER SAYS SO. No file, no record for yesterday, an
       unreadable file, a schema this build does not know, A DAY RECORD THAT IS
       NOT A RECORD, a record whose points are malformed - six different
       reasons, none of them a price, none of them an empty series, and the
       controller keeps its pre-FCST-04 plan through all of them. ⚠️ THE
       CRASHING ONE IS THE POINT: `{"days": {"2026-01-14": "oops"}}` loaded as
       healthy and then raised `AttributeError` out of every accessor, and both
       `PriceHistory(...)` and `.words` are unguarded in `Controller.__init__`
       - so the add-on did not start, and never self-healed.

THE 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/ uses. The horizon growth is a subtraction of two measured counts, not
    a constant. The cheapest-hour move is found by looking, not asserted.

  * NO STIMULUS IS EVALUATED WHERE THE QUANTITY IT GRADES IS DEGENERATE:
      - YESTERDAY AND TODAY MUST PRICE EVERY LOCAL HOUR DIFFERENTLY, or a fill
        that carried the wrong day is invisible. Asserted first, as a check.
      - THE PUBLISHED WINDOW MUST BE A STRICT SUBSET OF THE HORIZON, or "the
        fill widened it" and "the fill did nothing" are the same observation.
      - THE ELAPSED HOURS MUST CONTAIN THE CHEAPEST HOUR OF THE WHOLE FEED, or
        AC2 grades nothing: the windowed and unwindowed cheapest hour would be
        the same point either way. This is also the realistic shape of the
        danger - a cheap night that has already gone, holding a battery out of
        a trade all morning.
      - THE MERGE FIXTURE MUST REVISE A PRICE TO A DIFFERENT NUMBER, or merge
        and replace are indistinguishable.
      - THE PRUNE FIXTURE MUST HOLD MORE DAYS THAN KEEP_DAYS, or pruning is not
        exercised at all. ⚠️ THIS FILE BROKE ITS OWN RULE AND IT COST A LIVE
        MUTATION: §8's clock-jump check seeded ONE day, `_prune`'s early
        return fired, and the body it is named after never ran. The mutant that
        pruned by arithmetic against the clock - which empties the store on
        exactly the morning the fill is for - survived the whole 29-file suite.
        The fixture is now WRITTEN TO A FILE and reopened, because `_load()`
        does not prune and that is how a store legitimately holds more days
        than the cap.
      - NOW MUST NOT BE ON AN HOUR BOUNDARY. `datetime.now()` never is, and a
        horizon count taken at exactly :00:00 was green for a fill that emitted
        HORIZON + 1 hours at every other instant.
      - A SUB-HOURLY FIXTURE MUST REALLY BE SUB-HOURLY, or "every published
        print survives" passes for an hourly feed and grades nothing.

  * A FAILED READ IS NOT AN OBSERVATION. Every degraded store is graded on the
    TYPE it answers with (`no_data` and its reason), never on a falsy value,
    and the malformed-point case is checked for the specific defect - a price
    defaulted to 0.0 - rather than for a count.

  * AND A STRUCTURAL CHECK MUST BE ABLE TO FAIL. §3's ordering check compared
    `ast.walk` positions, which grade NESTING DEPTH: `self._remember(...)` is a
    statement and `self.yesterday(...)` is an argument of the fill, so
    `_remember` came out first whichever statement was written first - measured
    by reversing them and watching the check stay green, while DECISIONS §3 and
    the module docstring both cited it as making a reversal "red rather than
    plausible". It is now line numbers AND the behaviour itself.

⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or run
`tools/mutation_run.sh`, which does. See MUTATION.md and TEST-15.
"""

import ast
import asyncio
import inspect
import json
import os
import shutil
import sys
import tempfile
from datetime import datetime, timedelta, timezone

from app import price_history as _ph
from app.price import (
    Absence, PricePoint, PriceProduct, PriceRead, brussels,
)
from app.price_history import KEEP_DAYS, NoHistory, PriceHistory, SCHEMA
from app.static_tariff import resolve_tariff

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


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


def why(read):
    """A read's reason, or "" when it is a series.

    ⚠️ NOT `read.reason`. On a `_Series` that attribute RAISES `Absence` by
    design, so a mutation that makes a refusing accessor answer a series would
    take this file down with a traceback instead of reporting a FAIL - and
    MUTATION.md's rule is that "it was refused" and "it broke on the way to
    being refused" must not grade the same.
    """
    return read.match(series=lambda _r: "", no_data=lambda r: r.reason)


def raises(fn, exc):
    try:
        fn()
    except exc:
        return True
    except Exception:
        return False
    return False


class _Raised:
    """What a check saw instead of a value. Equals nothing, is falsy, and is
    NOT a default."""

    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`.

    ⚠️ NOT A DEFAULT, AND NOT A BARE CALL EITHER. MUTATION.md's rule is that
    "it was refused" and "it broke on the way to being refused" must not grade
    the same - and the accessors below are exactly the ones the unvalidated
    day record used to raise from, so a raw call here would take the harness
    down on the mutation it is meant to catch and report nothing at all.
    Copied from test_price_forecast.py.
    """
    try:
        return fn()
    except Exception as e:                                    # noqa: BLE001
        return _Raised(e)


def run(coro):
    return asyncio.run(coro)


def tmpdir():
    d = tempfile.mkdtemp(prefix="fcst04-")
    _tmpdirs.append(d)
    return d


def store():
    """A history on a fresh directory that looks durable."""
    return PriceHistory(os.path.join(tmpdir(), "price_history.json"))


# --------------------------------------------------------------------------- #
# The fixtures.
#
# ⚠️ A DAY IS A BRUSSELS DAY, BUILT BY SEARCH. The first draft of this file made
# a day 24 UTC hours long and every count in it came out one short: in January
# Belgium is UTC+1, so 23:00 UTC already belongs to tomorrow, and a "24-hour
# day" built by adding hours to midnight UTC is 23 hours of one Brussels day
# plus one hour of the next. That is the same class of error the module is
# keyed to avoid, so the fixture must not contain it - and it is found by
# LOOKING at each instant's Brussels date rather than by adding an offset a
# helper would have to be right about.
#
# ⚠️ THE INSTANT IS 09:00 BRUSSELS ON A WEEKDAY, WHICH IS THE WHOLE POINT. The
# day-ahead auction publishes at 14-16 h CET, so at this instant a healthy feed
# is serving TODAY IN FULL and nothing of tomorrow - the window this ticket
# exists for. A fixture at 20:00 would have both days and would grade nothing.
# --------------------------------------------------------------------------- #

TODAY_D = "2026-01-15"                           # a Thursday; Belgium on CET
YDAY_D = "2026-01-14"
HORIZON_H = 48


def bday_hours(bdate: str) -> list:
    """The UTC instants whose BRUSSELS date is `bdate`, one per local hour."""
    probe = datetime.fromisoformat(bdate + "T00:00:00+00:00") - timedelta(hours=3)
    out = []
    for n in range(30):
        u = probe + timedelta(hours=n)
        if brussels(u).strftime("%Y-%m-%d") == bdate:
            out.append(u)
    return out


TODAY_H = bday_hours(TODAY_D)
YDAY_H = bday_hours(YDAY_D)


def at_local(bdate: str, hh: int) -> datetime:
    """The UTC instant whose Brussels wall clock reads hh:00 on `bdate`."""
    return next(u for u in bday_hours(bdate) if brussels(u).hour == hh)


# ⚠️ AND IT IS NOT ON THE HOUR. A controller's `now` is
# `datetime.now(timezone.utc)`; it is :17:43 past as often as anything else and
# it is never :00:00. The first draft of this file used the exact hour, and
# §1's `_n_after == HORIZON_H` was green only because of it: `since` was
# floored to its hour while `until` was compared raw, so a 48 h request made at
# any other instant came back 49 hours long. A fixture has to have the shape
# production has, or a count grades an alignment the running system never gets.
NOW = at_local(TODAY_D, 9) + timedelta(minutes=17, seconds=43)

# Where the WINDOW starts: the whole hour that contains NOW. The hour a
# controller is standing in is still a tradeable hour, so the fill emits from
# here and not from NOW - and every count below is derived against this rather
# than against the instant.
SINCE_HOUR = NOW.replace(minute=0, second=0, microsecond=0)

# Today's feed, in EUR/MWh, keyed by BRUSSELS local hour and all distinct.
# ⚠️ THE CHEAPEST HOUR IS 03:00 LOCAL - ALREADY ELAPSED AT 09:00 - and that is
# AC2's stimulus, asserted below rather than assumed.
TODAY_MWH = {h: 120.0 + h for h in range(24)}
TODAY_MWH[3] = 5.0                       # the cheap night that has already gone
TODAY_MWH[9] = 250.0                     # the hour NOW falls in - dear, so the
                                         # cheapest hour AHEAD is strictly
                                         # later than NOW and "it moved to an
                                         # hour still ahead" is exact
TODAY_MWH[19] = 260.0                    # the dear evening still ahead

# Yesterday's published prices, DIFFERENT AT EVERY LOCAL HOUR from today's, so
# a fill that carried the wrong day is visible.
YESTERDAY_MWH = {h: 400.0 + 3 * h for h in range(24)}

MARKUP, TAXES = 0.03, 0.06


def allin(mwh):
    """EUR/MWh as published -> the all-in EUR/kWh `assemble()` produces."""
    return round(mwh * 0.001 + MARKUP + TAXES, 6)


def feed_body(instants, table):
    return json.dumps({"data": [
        {"start": u.isoformat(), "price_eur_mwh": table[brussels(u).hour]}
        for u in instants]}).encode()


def transport_for(body):
    async def _t(url, timeout_s):
        return body
    return _t


CFG = {"price_provider": "generic_dayahead",
       "price_feed_url": "https://api.energy-charts.info/price?bzn=BE",
       "price_timeout_s": 1,
       "price_markup_eur_kwh": MARKUP,
       "price_taxes_eur_kwh": TAXES,
       "static_profile": ""}


def series_of(instants, table, product=PriceProduct.DAY_AHEAD,
              provider="generic_dayahead", stamp=None):
    """A PriceRead.series built directly - the shape `load_prices()` returns."""
    return PriceRead.series(
        [PricePoint(start=brussels(u),
                    eur_per_kwh=allin(table[brussels(u).hour]),
                    provisional=False)
         for u in instants],
        product, provider, stamp or NOW)


# --------------------------------------------------------------------------- #
print("0. the arithmetic this file grades with must not be degenerate")

check(f"a Brussels day is 24 hours long in the fixture helper, on both fixture "
      f"days ({len(TODAY_H)}, {len(YDAY_H)}) - a 23-hour 'day' built by adding "
      f"to midnight UTC is what made every count below come out one short",
      len(TODAY_H) == 24 and len(YDAY_H) == 24
      and {brussels(u).hour for u in TODAY_H} == set(range(24)))

check(f"NOW is {NOW.isoformat()} - deliberately NOT on an hour boundary. A "
      f"horizon count taken at exactly :00:00 is green for a fill that emits "
      f"HORIZON + 1 hours at every instant a controller actually runs at",
      (NOW.minute, NOW.second, NOW.microsecond) != (0, 0, 0))

# The feed serves TODAY IN FULL - which is what a day-ahead feed does - so the
# hours it publishes that are still AHEAD are a strict subset of the horizon.
# ⚠️ AGAINST SINCE_HOUR, NOT NOW: the 09:00 point started before NOW and its
# hour still contains NOW, so it is a published hour of the window.
_ahead = [u for u in TODAY_H if u >= SINCE_HOUR]
check(f"at 09:00 the hours the feed published that have NOT elapsed are a "
      f"STRICT SUBSET of the {HORIZON_H} h horizon ({len(_ahead)} of "
      f"{HORIZON_H}), so 'the fill widened the plan' and 'the fill did "
      f"nothing' are different observations",
      0 < len(_ahead) < HORIZON_H)

check("yesterday prices EVERY local hour differently from today, so a fill "
      "that carried the wrong day is visible in the numbers",
      all(TODAY_MWH[h] != YESTERDAY_MWH[h] for h in range(24)))

check("...and no fixture price is zero, so 'dropped the price' is visible too",
      all(TODAY_MWH.values()) and all(YESTERDAY_MWH.values()))

_cheapest_all = min(TODAY_H, key=lambda u: TODAY_MWH[brussels(u).hour])
_cheapest_ahead = min(_ahead, key=lambda u: TODAY_MWH[brussels(u).hour])
check(f"AC2's stimulus is real: the cheapest hour of the whole published day is "
      f"{brussels(_cheapest_all).hour:02d}:00 local, which has already elapsed "
      f"at 09:00, and the cheapest hour still AHEAD is a different one "
      f"({brussels(_cheapest_ahead).hour:02d}:00). Without this the windowed "
      f"and unwindowed answers would be the same point and AC2 would grade "
      f"nothing",
      _cheapest_all != _cheapest_ahead and _cheapest_all < SINCE_HOUR
      and _cheapest_ahead > NOW)

check(f"KEEP_DAYS is {KEEP_DAYS}, and the prune fixture below files more days "
      f"than that, so pruning is exercised rather than skipped", KEEP_DAYS < 6)


# --------------------------------------------------------------------------- #
print()
print("1. AC1 - the hole: a partially-published feed now covers the horizon")

# ⚠️ THE FEED PUBLISHES TODAY IN FULL, WHICH IS WHAT A DAY-AHEAD FEED DOES.
# The partial publication this ticket is about is that TOMORROW is missing, not
# that this morning is - and the elapsed hours the feed is still serving are
# exactly what AC2 is about, so they have to be in the body.
_body_today = feed_body(TODAY_H, TODAY_MWH)
_until = NOW + timedelta(hours=HORIZON_H)

# The pre-FCST-04 behaviour, by the shipped code path: NoHistory() records
# nothing and carries nothing, so this IS the old plan.
_before = run(resolve_tariff(CFG, NOW, _until,
                             transport=transport_for(_body_today), now=NOW,
                             history=NoHistory()))

# And the same call with a history that already holds yesterday.
_h1 = store()
_h1_seed = series_of(YDAY_H, YESTERDAY_MWH)
_ph.PriceHistory._remember(_h1, _h1_seed, NOW)          # seeding, not a caller
_after = run(resolve_tariff(CFG, NOW, _until,
                            transport=transport_for(_body_today), now=NOW,
                            history=_h1))

_n_before = len(_before.read.points)
_n_after = len(_after.read.points)
check(f"the plan's horizon GREW, measured: {_n_before} h before FCST-04, "
      f"{_n_after} h after, against a {HORIZON_H} h request",
      _n_after > _n_before and _n_after == HORIZON_H)

check(f"...and before the change it was exactly what the feed sent "
      f"({len(TODAY_H)} h of one day), which is the defect from both ends: the "
      f"{HORIZON_H - len(_ahead)} hours past the feed's own horizon were not "
      f"short-priced but ABSENT, and the {len(TODAY_H) - len(_ahead)} hours "
      f"that had already elapsed were in the series anyway",
      _n_before == len(TODAY_H))

# The split, re-derived here from the fixture rather than read off the module.
_published_instants = {u.astimezone(UTC) for u in _ahead}
_got_carried = {p.start.astimezone(UTC) for p in _after.read.points
                if p.provisional}
_got_firm = {p.start.astimezone(UTC) for p in _after.read.points
             if not p.provisional}
check("AC1: exactly the hours the feed published are FIRM, and exactly the "
      "hours it did not are PROVISIONAL - the flag is a claim about "
      "provenance, re-derived here from the fixture",
      _got_firm == _published_instants
      and _got_carried == {p.start.astimezone(UTC)
                           for p in _after.read.points} - _published_instants
      and len(_got_carried) == HORIZON_H - len(_ahead))

_before_by_start = {p.start.astimezone(UTC): p for p in _before.read.points}
# ⚠️ THIS IS A VALUE CHECK AND SAYS SO. `_before` and `_after` come from two
# separate `resolve_tariff()` calls, each of which parsed the feed body into
# its OWN objects, so `is` could never hold between them however the fill were
# written - a check claiming identity here would be claiming a property its
# own fixture makes untestable. What it does grade is that the fill moved the
# WINDOW and did not touch a PRICE.
check("...and every published point carries the same NUMBER as the unfilled "
      "plan's for that instant: the fill changed which hours are in the "
      "series, not what any of them costs",
      all(p.eur_per_kwh == _before_by_start[p.start.astimezone(UTC)].eur_per_kwh
          for p in _after.read.points if not p.provisional))

# ...and identity itself, graded on a read this file constructed and still
# holds a reference to. FCST-03's whole anti-copy argument is
# `filled.points[i] is fetched.points[j]`, and floats cannot grade it.
_h1b = store()
_ph.PriceHistory._remember(_h1b, series_of(YDAY_H, YESTERDAY_MWH), NOW)
_read_id = series_of(TODAY_H, TODAY_MWH)
_filled_id = _h1b.carry(_read_id, since=NOW, until=_until, stamp=NOW)
_firm_id = [p for p in _filled_id.points if not p.provisional]
check(f"...and IDENTITY, graded as identity: all {len(_firm_id)} firm points "
      f"of a filled series ARE the provider's own objects (`is`), not copies "
      f"carrying the same numbers. A copy is where the provisional flag gets "
      f"lost (FCST-03)",
      len(_firm_id) == len(_ahead)
      and all(any(p is q for q in _read_id.points) for p in _firm_id))

# The carried numbers are yesterday's, at the same BRUSSELS wall-clock hour.
# ⚠️ THE LOOKUP IS BUILT BY SEARCH over yesterday's own instants, not by
# subtracting 24 h from today's - a fill that subtracted a fixed day is exactly
# what this is meant to catch, and a helper that subtracts one cannot catch it.
_Y_BY_LOCAL = {}
for _u in YDAY_H:
    _Y_BY_LOCAL.setdefault(brussels(_u).hour,
                           allin(YESTERDAY_MWH[brussels(_u).hour]))
_carried_ok = [_Y_BY_LOCAL.get(brussels(p.start).hour) is not None
               and abs(p.eur_per_kwh
                       - _Y_BY_LOCAL[brussels(p.start).hour]) < 1e-9
               for p in _after.read.points if p.provisional]
check(f"...and every carried number is YESTERDAY's price for the same BRUSSELS "
      f"wall-clock hour, not today's and not a neighbour's "
      f"({sum(_carried_ok)}/{len(_carried_ok)})",
      _carried_ok and all(_carried_ok))

# ⚠️ "FILLS EVERY UNPUBLISHED HOUR" IS CONDITIONAL, AND THIS IS THE CONDITION.
# A yesterday that is itself partial fills only the local hours it HOLDS; the
# rest are left out, which is §11.4 for those hours and emphatically not a
# zero. The behaviour is right and the wording was not - the CHANGELOG and DOCS
# said "every", unqualified, and nothing here exercised it.
_hp = store()
_PART_LOCAL = list(range(12))            # yesterday published 00:00..11:00 only
_ph.PriceHistory._remember(
    _hp, series_of([at_local(YDAY_D, h) for h in _PART_LOCAL], YESTERDAY_MWH),
    NOW)
_plan_part = run(resolve_tariff(CFG, NOW, _until,
                                transport=transport_for(_body_today), now=NOW,
                                history=_hp))
_window_hours = [SINCE_HOUR + timedelta(hours=n) for n in range(HORIZON_H)]
_ahead_utc = {u.astimezone(UTC) for u in _ahead}
_expect_covered = {u for u in _window_hours
                   if u in _ahead_utc or brussels(u).hour in _PART_LOCAL}
_got_covered = {p.start.astimezone(UTC) for p in _plan_part.read.points}
check(f"a PARTIAL yesterday fills only the local hours it holds: "
      f"{len(_got_covered)} of the {HORIZON_H} horizon hours are covered, "
      f"re-derived from the fixture as {len(_expect_covered)}. The hours "
      f"yesterday never published stay ABSENT",
      _got_covered == _expect_covered and 0 < len(_got_covered) < HORIZON_H)
_gap = sorted(set(_window_hours) - _expect_covered)[0]
check(f"...and an uncovered hour ({_gap.isoformat()}) raises `Absence` rather "
      f"than answering a price. §11.4 applies to that hour: a hole in the "
      f"series is not a free hour, and this is why the wording had to stop "
      f"saying 'every'",
      raises(lambda: _plan_part.read.at(_gap), Absence))

check("the plan reports itself PROVISIONAL now, which it did not before - the "
      "old plan claimed full firmness over a horizon it covered half of",
      _after.provisional and not _before.provisional)

check("...and the plan is still a provider plan, not a §11.4 fallback: the "
      "fill does not degrade anything",
      _after.origin == "provider" and _after.product is PriceProduct.DAY_AHEAD)


# --------------------------------------------------------------------------- #
print()
print("2. AC2 - the spread stopped being taken over hours that have gone")

_A = dict(peak_guard_active=False, min_spread_eur_kwh=0.0,
          uncertainty=frozenset())
_v_before = _before.arbitrage(**_A)
_v_after = _after.arbitrage(**_A)
check("both plans still ALLOW a trade, so this section compares two verdicts "
      "rather than a verdict and a refusal",
      _v_before.kind == "allowed" and _v_after.kind == "allowed")

_cheap_before = _v_before.cheapest.start.astimezone(UTC)
_cheap_after = _v_after.cheapest.start.astimezone(UTC)
check(f"AC2: the cheapest FIRM hour MOVED from {_cheap_before.isoformat()} - "
      f"already elapsed at {NOW.isoformat()}, so no cycle could ever buy in it "
      f"- to {_cheap_after.isoformat()}, which is still ahead",
      _cheap_before < NOW <= _cheap_after and _cheap_before != _cheap_after)

check("...and the new one is the cheapest hour the feed published that has NOT "
      "gone, derived from the fixture rather than from the module",
      _cheap_after == _cheapest_ahead.astimezone(UTC))

check("...and no point of the filled series starts before the requested "
      "`since` hour: the window is [since, until) and the past is not in it",
      min(p.start.astimezone(UTC) for p in _after.read.points) >= SINCE_HOUR)

# ⚠️ THE HONEST LIMIT OF THIS TICKET, checked so it cannot be overclaimed
# later: the CARRIED hours buy no trade. §8.3 forbids acting on a provisional
# price and refusal 3 drops them from the trade set, so what widened is what
# the optimiser can SEE, and the trade above came from the WINDOWING, not from
# yesterday's numbers.
_firm_after = [p for p in _after.read.points if not p.provisional]
check("the carried hours are NOT in the trade set - the verdict's cheapest and "
      "dearest are both firm, so this ticket widened what the optimiser sees "
      "and not what it may act on",
      not _v_after.cheapest.provisional and not _v_after.dearest.provisional
      and len(_firm_after) == len(_ahead))


# --------------------------------------------------------------------------- #
print()
print("2b. a QUARTER-HOURLY feed keeps every print, and its spread with them")

# ⚠️ THE DEFECT THIS SECTION EXISTS FOR. `eneco_eddy` is the Belgian imbalance
# product and settles per QUARTER; `_validate_raw()` accepts any
# strictly-increasing spacing, so a published hour may hold four points.
# FCST-04's first draft keyed the published set `{_utc_hour(p.start): p ...}` -
# LAST POINT OF THE HOUR WINS, the other three dropped - and on an all-firm,
# fully-published 96-point feed the plan came out 15 points long with its
# arbitrage spread down from 0.143 to 0.014 EUR/kWh. That is not the windowing
# effect AC2 is about; it is prints being thrown away, and no flag reports it.
_QSTEP_MIN = 15
_qrows = []
for _n in range(len(TODAY_H) * (60 // _QSTEP_MIN)):
    _qu = TODAY_H[0] + timedelta(minutes=_QSTEP_MIN * _n)
    # A spiky within-the-hour profile, so decimation is visible in the SPREAD
    # and not only in a count.
    _qrows.append({"start": _qu.isoformat(),
                   "price_eur_mwh": 100.0 + 40.0 * (_n % 4) + brussels(_qu).hour})
_qbody = json.dumps({"data": _qrows}).encode()
check(f"the fixture really is sub-hourly - {len(_qrows)} points across "
      f"{len(TODAY_H)} Brussels hours - or every check below passes for an "
      f"hourly feed too and grades nothing",
      len(_qrows) == 4 * len(TODAY_H)
      and len({_r["start"] for _r in _qrows}) == len(_qrows))

_q_before = run(resolve_tariff(CFG, NOW, _until, transport=transport_for(_qbody),
                               now=NOW, history=NoHistory()))
_hq = store()
_ph.PriceHistory._remember(_hq, series_of(YDAY_H, YESTERDAY_MWH), NOW)
_q_after = run(resolve_tariff(CFG, NOW, _until, transport=transport_for(_qbody),
                              now=NOW, history=_hq))
# What the feed published INSIDE the window, derived from the unfilled plan.
_q_win = [p for p in _q_before.read.points
          if p.start.astimezone(UTC) >= SINCE_HOUR]
_q_firm = [p for p in _q_after.read.points if not p.provisional]
check(f"every print the window contains survives the fill: {len(_q_firm)} firm "
      f"points against the {len(_q_win)} the feed published in that window. "
      f"One-per-hour keying leaves {len(_q_win) // 4}",
      len(_q_win) > 0 and len(_q_firm) == len(_q_win))
check("...and they are the SAME INSTANTS AND PRICES the feed published, so "
      "the count is not being made up by four hourly points wearing "
      "quarter-hourly starts",
      {(p.start.astimezone(UTC), p.eur_per_kwh) for p in _q_firm}
      == {(p.start.astimezone(UTC), p.eur_per_kwh) for p in _q_win})
check("...and no two of them share a whole hour by accident: the fixture puts "
      "four distinct prices inside every published hour, which is what the "
      "decimated version could not represent",
      len({p.start.astimezone(UTC).replace(minute=0) for p in _q_firm})
      == len(_q_firm) // 4 > 0)

_q_spread_want = round(max(p.eur_per_kwh for p in _q_win)
                       - min(p.eur_per_kwh for p in _q_win), 6)
_qv = _q_after.arbitrage(**_A)
_q_spread_got = _qv.match(allowed=lambda v: v.spread_eur_kwh,
                          refused=lambda _v: None)
check(f"...and the spread is the fixture's own max - min over that window "
      f"({_q_spread_want} EUR/kWh, measured {_q_spread_got}), not the "
      f"flattened figure a decimated series produces",
      _qv.kind == "allowed" and _q_spread_got is not None
      and abs(_q_spread_got - _q_spread_want) < 1e-9)
check("...and the cheapest firm point is one the FEED published, at its own "
      "sub-hourly instant - not a re-derived hourly one",
      _qv.kind == "allowed"
      and any(_qv.cheapest is q for q in _q_after.read.points)
      and any(_qv.cheapest.start.astimezone(UTC) == q.start.astimezone(UTC)
              for q in _q_win))


# --------------------------------------------------------------------------- #
print()
print("3. AC3 - the cache records the FEED, never its own output")

_h2 = store()
_read2 = series_of(TODAY_H, TODAY_MWH)
_filled2 = _h2.carry(_read2, since=NOW, until=_until, stamp=NOW)
# Round two, as the refresh task does every 1800 s: the same partial feed.
_filled2b = _h2.carry(_read2, since=NOW, until=_until, stamp=NOW)

_disk = json.load(open(_h2.path, encoding="utf-8"))
_today_key = brussels(NOW).strftime("%Y-%m-%d")
_recorded = {p["start"] for p in _disk["days"][_today_key]["points"]}
_recorded_utc = {datetime.fromisoformat(s).astimezone(UTC) for s in _recorded}
check(f"AC3: after two `carry()` rounds the record for {_today_key} holds "
      f"EXACTLY the {len(TODAY_H)} hours the feed published "
      f"({len(_recorded_utc)} held) - not one of the "
      f"{HORIZON_H - len(_ahead)} hours the fill invented",
      _recorded_utc == {u.astimezone(UTC) for u in TODAY_H})

check("...and no recorded point is flagged provisional, because a carried "
      "point never reached the file - the flag would still read true while the "
      "NUMBER decayed, which is the failure this rule is about",
      not any(p["provisional"] for p in _disk["days"][_today_key]["points"]))

def _canon(read):
    return [p.canonical() for p in read.match(series=lambda r: r.points,
                                              no_data=lambda _r: ())]


check(f"...and the fill is idempotent across the two rounds, which is §5.4's "
      f"late-publication contract - the refresh task runs it again every "
      f"1800 s against whatever the feed says by then "
      f"({len(_canon(_filled2))} h both times)",
      _canon(_filled2) and _canon(_filled2) == _canon(_filled2b))

# The ordering is the mechanism, so the mechanism is checked in the tree.
#
# ⚠️ EVERY SHIPPED PYTHON FILE, AND BY NAME RATHER THAN BY SUBSTRING. The first
# draft scanned `app/*.py` only - not `tools/`, not `scenarios/`, not the
# repo's `estop/` - and matched the bare string, so a module that merely
# MENTIONED `_remember` in a comment failed the check while a real call from
# `tools/` passed it unseen. This parses each file and looks for the name being
# USED: an attribute, a bare name, a def, or a string constant, which is how
# `getattr` reaches a private method.
_APP = os.path.join(HERE, "app")


def _uses_remember(path):
    """Does this module USE the name `_remember`? Prose does not count."""
    try:
        tree = ast.parse(open(path, encoding="utf-8").read())
    except (OSError, SyntaxError):
        # ⚠️ NOT False. A file this scan cannot read is a file it has not
        # cleared, and "unreadable" must never grade the same as "clean".
        return True
    for n in ast.walk(tree):
        if isinstance(n, ast.Attribute) and n.attr == "_remember":
            return True
        if isinstance(n, ast.Name) and n.id == "_remember":
            return True
        if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) \
                and n.name == "_remember":
            return True
        if isinstance(n, ast.Constant) and n.value == "_remember":
            return True
    return False


_SHIPPED = []
for _root in (_APP, os.path.join(HERE, "tools"), os.path.join(HERE, "scenarios"),
              os.path.join(os.path.dirname(HERE), "estop")):
    for _dirpath, _dirnames, _filenames in os.walk(_root):
        for _f in sorted(_filenames):
            if _f.endswith(".py") and _f != "price_history.py":
                _SHIPPED.append(os.path.join(_dirpath, _f))
check(f"the scan has a roster to walk ({len(_SHIPPED)} shipped modules across "
      f"app/, tools/, scenarios/ and estop/), and it reaches beyond app/ - an "
      f"empty or app-only glob passes for a leak in any other directory",
      len(_SHIPPED) > 10
      and any(os.sep + "tools" + os.sep in p for p in _SHIPPED))
_leaks = [os.path.relpath(p, HERE) for p in _SHIPPED if _uses_remember(p)]
check(f"AC3's mechanism is an ORDERING, not a validation - a carried point and "
      f"a published-provisional point are the same three fields - so nothing "
      f"outside app/price_history.py may name `_remember` {_leaks or 'none do'}",
      not _leaks)

_probe = os.path.join(tmpdir(), "probe.py")
open(_probe, "w", encoding="utf-8").write("def f(h):\n    return h._remember(1)\n")
_probe_ok = os.path.join(tmpdir(), "probe_ok.py")
open(_probe_ok, "w", encoding="utf-8").write(
    '"""A docstring about _remember."""\n# and a comment about _remember\nX = 1\n')
check("...and that scan FAILS on a real call and PASSES on a file that only "
      "talks about the rule, so it is a check and not a decoration",
      _uses_remember(_probe) and not _uses_remember(_probe_ok))

# ⚠️ `ast.walk` IS BREADTH-FIRST, SO A LIST OF CALL NAMES GRADES NESTING DEPTH
# AND NOT SOURCE ORDER. The first draft of this check was
# `_calls.index("_remember") < _calls.index("yesterday")`: `self._remember(...)`
# is a statement while `self.yesterday(...)` is an ARGUMENT of
# `fill_from_yesterday(...)`, one level deeper, so `_remember` was emitted
# first whichever statement came first. Measured by writing the reversal out
# and grading it: shipped and reversed both scored True, while DECISIONS §3 and
# price_history.py's docstring both cited the check as making a reversal "red
# rather than plausible". Line numbers here, and the behaviour itself below.
_carry_src = inspect.getsource(PriceHistory.carry)
_tree = ast.parse(_carry_src.strip())
_lines = {}
for _n in ast.walk(_tree):
    if isinstance(_n, ast.Call) and isinstance(_n.func, ast.Attribute):
        _lines[_n.func.attr] = min(_lines.get(_n.func.attr, 10 ** 9), _n.lineno)
check(f"`carry()` names `_remember` on an EARLIER LINE than `yesterday` "
      f"{_lines} - source order, which is what 'it records before it fills' "
      f"actually means",
      "_remember" in _lines and "yesterday" in _lines
      and _lines["_remember"] < _lines["yesterday"])

# ...and the BEHAVIOUR the ordering exists for, driven rather than read. A feed
# that hands back yesterday alongside today must refresh yesterday's record
# BEFORE it is consumed as the profile. Into an EMPTY store that is the whole
# difference: record-then-fill covers the horizon, fill-then-record covers
# nothing, because at the moment of the fill there is no yesterday yet.
_h_ord = store()                      # empty: no record from an earlier day
_two_day = PriceRead.series(
    [PricePoint(start=brussels(u),
                eur_per_kwh=allin(YESTERDAY_MWH[brussels(u).hour]),
                provisional=False) for u in YDAY_H]
    + [PricePoint(start=brussels(u),
                  eur_per_kwh=allin(TODAY_MWH[brussels(u).hour]),
                  provisional=False) for u in TODAY_H],
    PriceProduct.DAY_AHEAD, "generic_dayahead", NOW)
_ord_filled = _h_ord.carry(_two_day, since=NOW, until=_until, stamp=NOW)
_ord_n = _ord_filled.match(series=lambda r: len(r.points), no_data=lambda _r: 0)
check(f"...and the ordering is DRIVEN, not only read: a feed carrying "
      f"YESTERDAY alongside today, into an empty store, fills the whole "
      f"{HORIZON_H} h horizon ({_ord_n} h) because the record was written "
      f"before the lookup. Reversed, the lookup finds nothing and the plan is "
      f"the {len(_ahead)} published hours the window keeps",
      _ord_n == HORIZON_H)
check("...and the hours it carried carry YESTERDAY's numbers - the record that "
      "same call had just written - not today's",
      _ord_filled.kind == "series"
      and all(abs(p.eur_per_kwh
                  - allin(YESTERDAY_MWH[brussels(p.start).hour])) < 1e-9
              for p in _ord_filled.points if p.provisional))


# --------------------------------------------------------------------------- #
print()
print("4. rule 2 - 'yesterday' is a DATE, not 'the newest day I have'")

_h3 = store()
_ph.PriceHistory._remember(
    _h3, series_of(bday_hours("2026-01-12"), YESTERDAY_MWH), NOW)
_y3 = _h3.yesterday(NOW)
check("a store holding only the day before LAST answers no_data, not those "
      "prices - `newest()` is the accessor that decays, and an add-on down "
      "over a weekend is when it would",
      _y3.kind == "no_data")
check(f"...and the reason names BOTH the date wanted and the dates held, so an "
      f"operator can see which it is: {why(_y3)!r}",
      "2026-01-14" in why(_y3) and "2026-01-12" in why(_y3))

_h4 = store()
_ph.PriceHistory._remember(
    _h4, series_of(YDAY_H, YESTERDAY_MWH), NOW)
_y4 = _h4.yesterday(NOW)
check("...while the day BEFORE this one answers a series, so the refusal above "
      "is about the date and not about the store being broken",
      _y4.kind == "series" and len(_y4.points) == 24)

check("a naive instant is refused rather than assumed to be UTC - 'which "
      "Brussels day was yesterday' is the question a wall clock cannot answer",
      raises(lambda: _h4.yesterday(datetime(2026, 1, 15, 9)), ValueError))

# The Brussels-date rule, on the fixture a UTC-date rule gets wrong. A feed
# serving the 24 UTC hours of a July day spans TWO Brussels days - Belgium is
# on CEST, so 22:00 and 23:00 UTC are already 00:00 and 01:00 of tomorrow - and
# a UTC-keyed cache would file all 24 under one date. The evening hours are the
# expensive ones, so filing them under the wrong day is the costly direction.
_h5 = store()
_jul_utc = [datetime(2026, 7, 14, h, tzinfo=UTC) for h in range(24)]
_split = {brussels(u).strftime("%Y-%m-%d") for u in _jul_utc}
check(f"the July fixture really does straddle two Brussels days {sorted(_split)}"
      f", or the next check passes for a UTC-keyed cache as well",
      len(_split) == 2)
_ph.PriceHistory._remember(
    _h5, series_of(_jul_utc, YESTERDAY_MWH), _jul_utc[-1])
_disk5 = json.load(open(_h5.path, encoding="utf-8"))
_counts = {d: len(_disk5["days"][d]["points"]) for d in sorted(_disk5["days"])}
check(f"...and the record is keyed on the BRUSSELS date, so those 24 hours are "
      f"filed under both of them {_counts} rather than all under one",
      set(_counts) == _split and sum(_counts.values()) == 24
      and all(_counts.values()))


# --------------------------------------------------------------------------- #
print()
print("4b. the two mornings a year when 'yesterday' is a DATE and not 24 hours")

# ⚠️ `brussels()` RETURNS A FIXED-OFFSET DATETIME, so `brussels(stamp) -
# timedelta(days=1)` subtracts 24 ABSOLUTE HOURS from a wall clock. The
# Brussels day before a spring-forward is 23 hours long, so between 00:00 and
# 00:59 on the morning after it that lands on the day BEFORE yesterday: the
# fill then carries a two-day-old profile behind a flag whose only vocabulary
# is "this is a guess", or - once pruning has dropped that day - answers
# `no_data` naming a date that is not yesterday. Measured at
# 2026-03-30T00:30+02:00 on the shipped branch: it returned 2026-03-28's
# prices. DST is this module's headline claim and it had NO CHECK.
for _label, (_dm2, _dm1, _d0) in (
        ("spring forward - yesterday was 23 h long",
         ("2026-03-28", "2026-03-29", "2026-03-30")),
        ("fall back - yesterday was 25 h long",
         ("2026-10-24", "2026-10-25", "2026-10-26"))):
    _lens = [len(bday_hours(_d)) for _d in (_dm2, _dm1, _d0)]
    check(f"{_label}: the fixture days are {_lens} hours long, so one of them "
          f"really is not 24 and a DATE subtraction and an INSTANT subtraction "
          f"can be told apart at all",
          _lens[1] != 24 and _lens[0] == 24 and _lens[2] == 24)
    _hd = store()
    _tag = {}
    for _i, _d in enumerate((_dm2, _dm1, _d0)):
        # One price band per DAY, so the answer names which day it came from.
        _tag[_d] = 0.100000 * (_i + 1)
        _ph.PriceHistory._remember(
            _hd,
            PriceRead.series(
                [PricePoint(start=brussels(u),
                            eur_per_kwh=round(_tag[_d] + 0.000100
                                              * brussels(u).hour, 6),
                            provisional=False)
                 for u in bday_hours(_d)],
                PriceProduct.DAY_AHEAD, "generic_dayahead", at_local(_d, 12)),
            at_local(_d, 12))
    check(f"{_label}: all three days are held, so the check below is about the "
          f"DATE arithmetic and not about a record having been pruned away",
          sorted(_hd._days) == sorted((_dm2, _dm1, _d0)))
    _stamp_dst = at_local(_d0, 0) + timedelta(minutes=30)
    _yd = _hd.yesterday(_stamp_dst)
    _band = _yd.match(
        series=lambda r: round(min(p.eur_per_kwh for p in r.points), 4),
        no_data=lambda _r: None)
    check(f"{_label}: at 00:30 local on {_d0}, yesterday() answers {_dm1}'s "
          f"prices (band {_band}, wanted {round(_tag[_dm1], 4)}) and not "
          f"{_dm2}'s (band {round(_tag[_dm2], 4)}) {why(_yd)!r}",
          _yd.kind == "series" and _band == round(_tag[_dm1], 4))
    check(f"{_label}: ...and it holds {_lens[1]} points, the length that "
          f"Brussels day actually was - a 24-slot answer would be the wrong "
          f"day wearing the right date",
          _yd.kind == "series" and len(_yd.points) == _lens[1])
    # The control, one hour later: an implementation that is right only at
    # 01:30 is the one being caught, so both are graded.
    _yd2 = _hd.yesterday(at_local(_d0, 1) + timedelta(minutes=30))
    check(f"{_label}: ...and at 01:30 local it is still {_dm1} - the defect "
          f"shows only in the first local hour, which is why an hourly refresh "
          f"could hit it and never be noticed",
          _yd2.kind == "series"
          and round(min(p.eur_per_kwh for p in _yd2.points), 4)
          == round(_tag[_dm1], 4))


# --------------------------------------------------------------------------- #
print()
print("5. AC4 - degraded stores, one reason each, no prices")

_missing = PriceHistory(os.path.join(tmpdir(), "gone", "price_history.json"))
check("no durable volume: the store is NOT ok, and it does not create the "
      "directory - a cache on the ephemeral layer reads fine and is thrown "
      "away on the next restart (tariff.PeakStore's rule)",
      not _missing.ok and "no durable volume" in (_missing.last_error or "")
      and not os.path.isdir(os.path.dirname(_missing.path)))

# ⚠️ AND IT DOES NOT FILL IN MEMORY EITHER. FCST-04 shipped this half-done:
# `_save()` short-circuited on `not ok` while `_remember()` populated
# `self._days` anyway, so a site with no /data answered `words` = "unavailable"
# and `yesterday()` = A SERIES - an in-memory profile that works all day and is
# gone on the next restart. price_history.py's own third warning rejects
# exactly that ("a yesterday that silently forgets") as its reason for not
# calling makedirs, so the code and the docstring have to agree.
_carried_nowhere = _missing.carry(series_of(TODAY_H, TODAY_MWH),
                                  since=NOW, until=_until, stamp=NOW)
check("a store with no durable volume records NOTHING, not even in memory: "
      "after a full carry it still holds no day, `yesterday()` still answers "
      "no_data, and `words` still says unavailable. A cache that forgets on "
      "restart without saying so is worse than one that says it cannot answer",
      _missing._days == {}
      and _missing.yesterday(NOW + timedelta(days=1)).kind == "no_data"
      and _missing.words.startswith("unavailable"))
check("...and the carry still returns a real series - the fill windows the "
      "feed and carries nothing, which is the pre-FCST-04 horizon and not a "
      "degradation",
      _carried_nowhere.kind == "series"
      and len(_carried_nowhere.points) == len(_ahead))

_d6 = tmpdir()
_p6 = os.path.join(_d6, "price_history.json")
open(_p6, "w", encoding="utf-8").write("{ this is not json")
_h6 = PriceHistory(_p6)
check("an unreadable file: the reason is kept and reported, and the record is "
      "NOT silently re-created as an empty one that reads like a first day",
      "unreadable" in (_h6.last_error or "")
      and _h6.yesterday(NOW).kind == "no_data"
      and "unreadable" in why(_h6.yesterday(NOW)))

_p7 = os.path.join(tmpdir(), "price_history.json")
json.dump({"version": SCHEMA + 99, "days": {}}, open(_p7, "w", encoding="utf-8"))
_h7 = PriceHistory(_p7)
check("a file from a schema this build does not know is discarded rather than "
      "guessed at - its points might mean something else",
      "schema" in (_h7.last_error or "")
      and _h7.yesterday(NOW).kind == "no_data")

# A record whose points are malformed. THE DEFECT IS A DEFAULTED PRICE.
_p8 = os.path.join(tmpdir(), "price_history.json")
_good = {"start": brussels(at_local(YDAY_D, 5)).isoformat(),
         "eur_per_kwh": allin(YESTERDAY_MWH[5]), "provisional": False}
json.dump({"version": SCHEMA, "days": {"2026-01-14": {
    "product": "dynamic_dayahead", "provider": "generic_dayahead",
    "points": [_good,
               {"start": "2026-01-14T06:00:00+01:00"},          # no price
               {"start": "2026-01-14 07:00:00", "eur_per_kwh": 0.2,
                "provisional": False},                          # naive
               {"eur_per_kwh": 0.3, "provisional": False},       # no instant
               "not even a dict"]}}},
          open(_p8, "w", encoding="utf-8"))
_y8 = PriceHistory(_p8).yesterday(NOW)
check("a malformed point is DROPPED, not defaulted: the four broken rows are "
      "gone and the one good row survives",
      _y8.kind == "series" and len(_y8.points) == 1)
check("...and specifically no point came back priced 0.0 - "
      "`float(row.get('eur_per_kwh', 0))` is how a lost field becomes 'that "
      "hour was free'",
      all(p.eur_per_kwh != 0.0 for p in _y8.points)
      and abs(_y8.points[0].eur_per_kwh - allin(YESTERDAY_MWH[5])) < 1e-9)

_p9 = os.path.join(tmpdir(), "price_history.json")
json.dump({"version": SCHEMA, "days": {"2026-01-14": {
    "product": "no_such_product", "provider": "x",
    "points": [_good]}}}, open(_p9, "w", encoding="utf-8"))
check("a record naming a product that is not a PriceProduct is no_data, not a "
      "series with a string in the field the optimiser switches on",
      PriceHistory(_p9).yesterday(NOW).kind == "no_data")

# ⚠️ AND THE ONE THAT STOPPED THE ADD-ON FROM STARTING. `_load()` validated
# `version` and that `days` is a dict, and never that each VALUE is one. With
# `{"version": 1, "days": {"2026-01-14": "oops"}}` the store reported ok=True
# and last_error=None, and then `rec.get(...)` raised `AttributeError` out of
# `words`, `yesterday()` and `resolve_tariff()` alike. `PriceHistory(...)` and
# `.words` are both unguarded in `Controller.__init__`, so the controller did
# not come up - and it never self-healed, because `_remember()` raised on the
# same record before it could reach `_save()`. The asymmetry that produced it
# is visible in the file: `_parse_points()` is carefully defensive about every
# field of a POINT, and `_load()` was not defensive about the record at all.
_p14 = os.path.join(tmpdir(), "price_history.json")
json.dump({"version": SCHEMA, "days": {
    "2026-01-14": "oops",
    "2026-01-13": {"product": "dynamic_dayahead", "provider": "generic_dayahead",
                   "points": [_good]}}},
          open(_p14, "w", encoding="utf-8"))
_h14 = PriceHistory(_p14)
check("a day record that is not a dict is DROPPED and REPORTED, and the rest "
      "of the file still loads - one bad record is not a bad file",
      _h14.ok and "unusable day record" in (_h14.last_error or "")
      and sorted(_h14._days) == ["2026-01-13"])
_w14 = derived(lambda: _h14.words)
_y14 = derived(lambda: _h14.yesterday(NOW).kind)
_plan14 = derived(lambda: run(resolve_tariff(
    CFG, NOW, _until, transport=transport_for(_body_today),
    now=NOW, history=_h14)).origin)
check(f"...and NOTHING raises on it: not `words` ({_w14!r}), not `yesterday()` "
      f"({_y14!r}), not the shipped `resolve_tariff()` path ({_plan14!r}). A "
      f"broken cache is a short horizon, not a controller that will not start",
      isinstance(_w14, str) and _y14 == "no_data" and _plan14 == "provider")
for _shape in ("oops", [1, 2, 3], 7, None, {"points": "not a list"}):
    _pX = os.path.join(tmpdir(), "price_history.json")
    json.dump({"version": SCHEMA, "days": {"2026-01-14": _shape}},
              open(_pX, "w", encoding="utf-8"))
    _hX = PriceHistory(_pX)
    _wX = derived(lambda h=_hX: h.words)
    _yX = derived(lambda h=_hX: h.yesterday(NOW).kind)
    check(f"...and the same holds for a day record of {type(_shape).__name__} "
          f"{_shape!r}: dropped, reported, and every accessor answers "
          f"({_wX!r}, {_yX!r})",
          _hX._days == {} and "unusable day record" in (_hX.last_error or "")
          and isinstance(_wX, str) and _yX == "no_data")

_pF = os.path.join(tmpdir(), "price_history.json")
json.dump({"version": float(SCHEMA), "days": {}},
          open(_pF, "w", encoding="utf-8"))
check(f"a `version` of {float(SCHEMA)!r} is refused although it EQUALS "
      f"{SCHEMA} - a schema check that cannot tell 1 from 1.0 is one that "
      f"would not notice a writer this build has never met",
      "schema" in (PriceHistory(_pF).last_error or ""))


# --------------------------------------------------------------------------- #
print()
print("5b. `words` - the status row, and one of the two things a bad record "
      "used to crash")

def words_of(h):
    return derived(lambda: h.words)


_w_missing = words_of(_missing)
check(f"no durable volume: the row says UNAVAILABLE and names the reason "
      f"{_w_missing!r}",
      isinstance(_w_missing, str) and _w_missing.startswith("unavailable")
      and "no durable volume" in _w_missing)
_h_words = store()
check("a healthy but EMPTY store says something different, because it is a "
      "different state and a normal one on a site's first day",
      isinstance(words_of(_h_words), str)
      and words_of(_h_words).startswith("empty")
      and "nothing to carry forward" in words_of(_h_words))
_ph.PriceHistory._remember(_h_words, series_of(YDAY_H, YESTERDAY_MWH), NOW)
check(f"...and a store holding a day names the DATE and the HOUR COUNT "
      f"{words_of(_h_words)!r} - 'a cache exists' and 'the cache holds "
      f"yesterday, all 24 hours of it' are different facts, and only the "
      f"second says the horizon will be full tomorrow morning",
      words_of(_h_words) == f"{YDAY_D} (24 h)")
check(f"a file that only PARTLY loaded says so in the row {_w14!r}, or the "
      f"days it dropped read as days that were simply never written",
      isinstance(_w14, str) and (_h14.last_error or "") in _w14
      and "2026-01-13" in _w14)
check("NoHistory() has a sentence of its own rather than an empty cell",
      NoHistory().words == "not kept")
check("...and `words` never raises, whatever the file holds. It is read once "
      "from Controller.__init__ and again on every status render, and neither "
      "call site guards it - an exception here is an add-on that does not start",
      all(isinstance(words_of(h), str)
          for h in (_missing, _h6, _h7, _h14, _h_words)))

# ⚠️ AND A FAILED WRITE LEAVES NOTHING BEHIND. `_save()` writes to a `mkstemp`
# and renames - atomic, so a power cut cannot leave a truncated file that
# parses as an empty day - and a failure BETWEEN the two used to leave the temp
# in place. The refresh task writes 48 times a day, so a persistent write
# failure leaked 48 inodes a day onto the volume whose being full is the
# likeliest reason the write failed at all.
_dleak = tmpdir()
_pleak = os.path.join(_dleak, "price_history.json")
_hleak = PriceHistory(_pleak)
check("the leak fixture starts clean, or 'nothing left behind' is not a "
      "measurement",
      _hleak.ok and os.listdir(_dleak) == [])
os.mkdir(_pleak)              # a DIRECTORY where the file goes: the temp is
                              # written fine and `os.replace` cannot complete
_ph.PriceHistory._remember(_hleak, series_of(YDAY_H, YESTERDAY_MWH), NOW)
check(f"a write that cannot complete SAYS so rather than passing quietly "
      f"{(_hleak.last_error or '')[:48]!r}",
      "could not write" in (_hleak.last_error or ""))
_left = sorted(n for n in os.listdir(_dleak) if n != "price_history.json")
check(f"...and leaves NOTHING behind {_left} - at 48 writes a day a leaked "
      f"temp per write fills the volume that was already the problem",
      _left == [])
_dok = tmpdir()
_hok = PriceHistory(os.path.join(_dok, "price_history.json"))
_ph.PriceHistory._remember(_hok, series_of(YDAY_H, YESTERDAY_MWH), NOW)
check(f"...and a SUCCESSFUL write leaves exactly the one file "
      f"{sorted(os.listdir(_dok))}, so the cleanup did not start deleting the "
      f"cache it just wrote",
      sorted(os.listdir(_dok)) == ["price_history.json"]
      and _hok.last_error is None)

# ...and every one of those leaves the CONTROLLER regulating.
#
# ⚠️ THE PLAN IS NOT BIT-FOR-BIT THE PRE-FCST-04 ONE, and the first draft of
# this check claimed it was. A broken cache still goes through the fill, which
# WINDOWS to [since, until) whether or not it has anything to carry - so the
# plan is the feed's own hours MINUS the elapsed ones. That is the AC2 half of
# this change, and it is correct with no cache at all: an hour that has gone is
# not a cheap hour. What must survive a broken cache is that the plan is a real
# provider series with no invented points and no §11.4 degradation.
for _label, _hist in (("no durable volume", _missing),
                      ("unreadable file", _h6),
                      ("wrong schema", _h7)):
    _plan = run(resolve_tariff(CFG, NOW, _until,
                               transport=transport_for(_body_today), now=NOW,
                               history=_hist))
    check(f"...and with {_label} the plan is a real provider series over the "
          f"{len(_ahead)} published hours still ahead, with NOTHING carried "
          f"and no §11.4 degradation. A broken cache is not a broken "
          f"controller - it is a short horizon, which is where this started",
          _plan.origin == "provider"
          and len(_plan.read.points) == len(_ahead)
          and not any(p.provisional for p in _plan.read.points))


# --------------------------------------------------------------------------- #
print()
print("6. rule 4 - hours accumulate within a day; they do not replace it")

_h10 = store()
_ph.PriceHistory._remember(_h10, series_of(TODAY_H, TODAY_MWH), NOW)
# A feed that serves only the hours left in the day - a legitimate shape, and
# the one that turns replace-the-record into a two-hour yesterday.
_REVISED = dict(TODAY_MWH)
_REVISED[22], _REVISED[23] = 999.0, 998.0
check("the revision really changes the numbers, or merge and replace are the "
      "same observation",
      _REVISED[22] != TODAY_MWH[22] and _REVISED[23] != TODAY_MWH[23])
_ph.PriceHistory._remember(
    _h10, series_of([at_local(TODAY_D, 22), at_local(TODAY_D, 23)], _REVISED),
    NOW)

_back = _h10.yesterday(NOW + timedelta(days=1))
check(f"a two-hour read did not overwrite a 24-hour day: "
      f"{len(_back.points) if _back.kind == 'series' else _back.kind} hours "
      f"held. Under replace, the next morning's 'yesterday' would have been "
      f"two hours long",
      _back.kind == "series" and len(_back.points) == 24)

# ⚠️ KEYED BY BRUSSELS HOUR, because that is what the fixture tables are keyed
# by. Keying this by UTC hour reads the wrong row of the table in winter and
# grades the module against a price nobody put there.
_by_hour = {brussels(p.start).hour: p.eur_per_kwh for p in _back.points}


def _priced(h, want):
    """True only if hour `h` is present AND carries `want`. A MISSING hour is
    False, never a KeyError: under `replace` the hours simply are not there, and
    a harness that dies on the mutation it is grading reports nothing at all."""
    return h in _by_hour and abs(_by_hour[h] - want) < 1e-9


check("...and the two revised hours carry the NEW price, because a later read "
      "is the newer statement about the same hour",
      _priced(22, allin(999.0)) and _priced(23, allin(998.0)))
check("...while every untouched hour kept its original price",
      all(_priced(h, allin(TODAY_MWH[h])) for h in range(22)))


# --------------------------------------------------------------------------- #
print()
print("7. the restart this whole module exists for")

_d11 = tmpdir()
_p11 = os.path.join(_d11, "price_history.json")
_first = PriceHistory(_p11)
_ph.PriceHistory._remember(
    _first, series_of(YDAY_H, YESTERDAY_MWH), NOW)
del _first
_second = PriceHistory(_p11)                    # a brand-new add-on process
_y11 = _second.yesterday(NOW)
check("yesterday survives a restart, which is the ONLY reason this is a file "
      "and not a variable on the refresh task: the window the fill exists for "
      "is midnight -> publication, and an add-on restarted at 08:00 would "
      "otherwise spend all of it on the truncated series",
      _y11.kind == "series" and len(_y11.points) == 24)
check("...with the prices intact through the JSON round trip, to six decimals",
      all(abs(p.eur_per_kwh
              - allin(YESTERDAY_MWH[brussels(p.start).hour])) < 1e-9
          for p in _y11.points))

_plan11 = run(resolve_tariff(CFG, NOW, _until,
                             transport=transport_for(_body_today), now=NOW,
                             history=_second))
check("...and the restarted controller's very first refresh already covers the "
      "whole horizon",
      len(_plan11.read.points) == HORIZON_H)


# --------------------------------------------------------------------------- #
print()
print("8. pruning keeps yesterday, and a jumped clock does not empty the store")

_h12 = store()
for back in range(6, 0, -1):
    _ph.PriceHistory._remember(
        _h12,
        series_of(bday_hours(
            (datetime.fromisoformat(TODAY_D) - timedelta(days=back)
             ).strftime("%Y-%m-%d")), YESTERDAY_MWH),
        NOW)
_held = sorted(json.load(open(_h12.path, encoding="utf-8"))["days"])
check(f"six days filed, {KEEP_DAYS} held {_held} - a bigger cache would only "
      f"hold days rule 2 refuses to look at",
      len(_held) == KEEP_DAYS)
check("...and yesterday is one of them, which is the only day that matters",
      _h12.yesterday(NOW).kind == "series")

# ⚠️ THE STIMULUS HAS TO REACH `_prune`'s BODY, AND THE FIRST DRAFT DID NOT.
# It seeded ONE day and then pruned with a jumped stamp: `if len(self._days) <=
# KEEP_DAYS: return` fired, the body never ran, and the check named after that
# body was green for an implementation that pruned by clock arithmetic.
# Measured - the body was replaced with "delete every day older than stamp
# minus KEEP_DAYS days", the guard left in place, and the whole 29-file suite
# stayed green while the mutant emptied the store on a jumped clock.
#
# A store holding more days than the cap is a REAL state, not a contrivance:
# `_load()` does not prune, so this is exactly what a restart sees after a file
# written by a build with a larger cap, or after a feed that spanned several
# days. So the fixture is written, opened, and only then pruned.
_FILED = KEEP_DAYS + 2
_p13 = os.path.join(tmpdir(), "price_history.json")
_days13 = {}
for _back in range(_FILED - 1, -1, -1):
    _d13 = (datetime.fromisoformat(TODAY_D)
            - timedelta(days=_back)).strftime("%Y-%m-%d")
    _days13[_d13] = {
        "product": PriceProduct.DAY_AHEAD.value,
        "provider": "generic_dayahead",
        "points": [PricePoint(start=brussels(u),
                              eur_per_kwh=allin(YESTERDAY_MWH[brussels(u).hour]),
                              provisional=False).canonical()
                   for u in bday_hours(_d13)]}
json.dump({"version": SCHEMA, "days": _days13},
          open(_p13, "w", encoding="utf-8"))
_h13 = PriceHistory(_p13)
check(f"the prune fixture holds {len(_h13._days)} days, MORE than KEEP_DAYS "
      f"({KEEP_DAYS}), so the early return cannot swallow the stimulus - this "
      f"file's own rule, and the one its first draft broke",
      len(_h13._days) == _FILED > KEEP_DAYS)
_ph.PriceHistory._prune(_h13, NOW + timedelta(days=4000))
check(f"pruning does not delete by ARITHMETIC AGAINST THE CLOCK: a clock that "
      f"jumped four thousand days keeps the newest {KEEP_DAYS} of the "
      f"{_FILED} records {sorted(_h13._days)} instead of deleting all of them, "
      f"which would empty the store on exactly the morning the fill is for",
      len(_h13._days) == KEEP_DAYS and _h13.yesterday(NOW).kind == "series")


# --------------------------------------------------------------------------- #
print()
print("8b. the two days a lookup can ask for are PINNED, so a far-off record "
      "cannot evict yesterday")

# ⚠️ KEEP_DAYS = 3 IS THE STEADY STATE EXACTLY - {yesterday, today, tomorrow} -
# so a newest-N cap has NO SLACK. A feed that publishes to D+2 makes four days
# and the cap drops the oldest, which is yesterday: the only day the fill can
# use. One stray point in the wrong year does the same thing permanently,
# because a lexicographic sort puts it LAST and nothing can ever age it out.
_h15 = store()
_SPAN_D = (YDAY_D, TODAY_D, "2026-01-16", "2026-01-17")
_span = [u for _d in _SPAN_D for u in bday_hours(_d)]
check(f"the spanning fixture really covers {len(_SPAN_D)} Brussels days, more "
      f"than KEEP_DAYS ({KEEP_DAYS}), or the cap is not under pressure and "
      f"this grades nothing",
      len(_SPAN_D) > KEEP_DAYS
      and len({brussels(u).strftime("%Y-%m-%d") for u in _span}) == len(_SPAN_D))
_ph.PriceHistory._remember(_h15, series_of(_span, YESTERDAY_MWH), NOW)
check(f"a feed spanning {len(_SPAN_D)} days does NOT evict yesterday: "
      f"{sorted(_h15._days)} held, and {YDAY_D} is one of them. Under the "
      f"newest-{KEEP_DAYS} cap alone it is the first record dropped, and the "
      f"fill loses its only usable day for good",
      YDAY_D in _h15._days and _h15.yesterday(NOW).kind == "series")

_h16 = store()
for _d16 in (YDAY_D, TODAY_D, "2026-01-16"):
    _ph.PriceHistory._remember(_h16, series_of(bday_hours(_d16), TODAY_MWH), NOW)
_STRAY_D = "2099-07-01"
_ph.PriceHistory._remember(
    _h16, series_of(bday_hours(_STRAY_D)[:1], TODAY_MWH), NOW)
check(f"the stray fixture is really far future and really sorts last "
      f"{sorted(_h16._days)[-1]!r}, or 'it can never age out' is not what is "
      f"being graded",
      sorted(_h16._days)[-1] == _STRAY_D)
check(f"...and ONE stray point in {_STRAY_D} does not push yesterday out "
      f"either. It is not deleted - nothing is deleted by arithmetic here - "
      f"it simply cannot take yesterday's slot",
      YDAY_D in _h16._days and _h16.yesterday(NOW).kind == "series")
check(f"...while the store stays BOUNDED at KEEP_DAYS + 2 = {KEEP_DAYS + 2} "
      f"records ({len(_h16._days)} held), so the pin is a pin and not a leak",
      len(_h16._days) <= KEEP_DAYS + 2)


# --------------------------------------------------------------------------- #
print()
print("9. the parameter cannot be forgotten, and main.py does not forget it")

_sig = inspect.signature(resolve_tariff)
_hp = _sig.parameters.get("history")
check("`resolve_tariff(history=...)` is KEYWORD-ONLY and has NO DEFAULT - a "
      "default of None meaning 'skip the fill' would hand a caller who forgot "
      "it the truncated horizon this ticket removes, silently",
      _hp is not None and _hp.kind is inspect.Parameter.KEYWORD_ONLY
      and _hp.default is inspect.Parameter.empty)

check("...so calling it without one is a TypeError, not a quiet regression",
      raises(lambda: run(resolve_tariff(CFG, NOW, _until,
                                        transport=transport_for(_body_today),
                                        now=NOW)), TypeError))

_MAIN = open(os.path.join(_APP, "main.py"), encoding="utf-8").read()
_main_tree = ast.parse(_MAIN)
_refresh = next(n for n in ast.walk(_main_tree)
                if isinstance(n, ast.AsyncFunctionDef)
                and n.name == "run_tariff_refresh")
_kw = [k.arg for n in ast.walk(_refresh) if isinstance(n, ast.Call)
       for k in n.keywords]
check(f"the shipped refresh task passes a history {_kw} - the wiring is what "
      f"this ticket IS, so it is checked in the tree rather than described",
      "history" in _kw)
check("the FCST-04 note in main.py carries the WINDOWING caveat beside the "
      "'widens what the optimiser can SEE, never what it may act on' "
      "guarantee. That note is the one place a maintainer reads that sentence "
      "unqualified, and the CHANGELOG, DOCS and resolve_tariff()'s docstring "
      "all carry the pair - so it may not appear here on its own",
      "never what it may act on" in _MAIN and "ALREADY ELAPSED" in _MAIN)

check("...and it is the CONTROLLER'S store, not a NoHistory() smuggled in to "
      "make the call typecheck",
      "history=self.price_history" in _MAIN
      and "NoHistory" not in _MAIN)

check("...and `words` reaches the STATUS PAGE as its own row, so what the fill "
      "has to carry from is a fact an operator can read rather than one that "
      "only exists in a log line",
      '("Price history", self.price_history.words)' in _MAIN)
_rows_fn = next(n for n in ast.walk(_main_tree)
                if isinstance(n, ast.FunctionDef)
                and n.name == "status_payload")
_row_labels = [e.elts[0].value for e in ast.walk(_rows_fn)
               if isinstance(e, ast.Tuple) and len(e.elts) == 2
               and isinstance(e.elts[0], ast.Constant)
               and isinstance(e.elts[0].value, str)]
check(f"...and it is in the ROW LIST rather than in a comment about one "
      f"{[r for r in _row_labels if 'rice' in r or 'ariff' in r]}",
      "Price history" in _row_labels)

# ⚠️ AND WHAT THE STATUS PAGE DOES **NOT** SAY, PINNED AS A PAIR. DOCS.md
# claimed "*Tariff* on the status page reads `provisional` through the
# morning". It does not and never did: `self.tariff_words` is assigned only
# from the Home Assistant price-entity branch and from `static_status()`, both
# on the 30 s tick, and the provider plan's own sentence reaches `_LOG.info`
# and nothing else. This grades the PAIRING rather than the current state - so
# wiring the plan into that row is fine and correcting the doc is fine, and
# doing one without the other is red.
_MAIN_LINES = _MAIN.splitlines()
_tw_lines = sorted({n.lineno for n in ast.walk(_main_tree)
                    if isinstance(n, ast.Assign)
                    for t in n.targets
                    for a in ast.walk(t)
                    if isinstance(a, ast.Attribute) and a.attr == "tariff_words"})
_row_from_plan = any("plan" in _MAIN_LINES[i - 1] for i in _tw_lines)
_DOCS = open(os.path.join(HERE, "DOCS.md"), encoding="utf-8").read()
_docs_claims = "**Tariff** on the status page reads `provisional`" in _DOCS
check(f"the *Tariff* row and what DOCS says about it AGREE: fed from the "
      f"provider plan = {_row_from_plan} (assigned on lines {_tw_lines}), DOCS "
      f"claims it reads `provisional` through the morning = {_docs_claims}. It "
      f"shipped as False against True - an overclaim in a doc is a defect "
      f"here, and this is the pair that has to move together",
      len(_tw_lines) >= 2 and _row_from_plan == _docs_claims)

# NoHistory is inert, and says so by identity rather than by producing a series.
_nh_read = series_of(TODAY_H, TODAY_MWH)
check("NoHistory() carries nothing and returns the read BY IDENTITY, so the "
      "explicit 'I keep no history' is exactly the old behaviour and not a "
      "second implementation of it",
      NoHistory().carry(_nh_read, since=NOW, until=_until, stamp=NOW)
      is _nh_read
      and NoHistory().yesterday(NOW).kind == "no_data")


# --------------------------------------------------------------------------- #
for d in _tmpdirs:
    shutil.rmtree(d, ignore_errors=True)

print()
if fails:
    print(f"{len(fails)} FAILED: " + "; ".join(fails))
    sys.exit(1)
print("all checks passed")