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 record whose points are malformed - four different
       reasons, none of them a price, none of them an empty series, and the
       controller keeps its pre-FCST-04 plan through all four.

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.

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

⚠️ 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 (
    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


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)


NOW = at_local(TODAY_D, 9)

# 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[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)))

# 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.
_ahead = [u for u in TODAY_H if u >= NOW]
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 < NOW
      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}
check("...and every published point is the PROVIDER'S OWN OBJECT, passed "
      "through by identity - a copy is where the flag gets lost (FCST-03)",
      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))

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

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)
      >= NOW.replace(minute=0, second=0, microsecond=0))

# ⚠️ 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("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.
_APP = os.path.join(HERE, "app")
_leaks = []
for name in sorted(os.listdir(_APP)):
    if not name.endswith(".py") or name == "price_history.py":
        continue
    src = open(os.path.join(_APP, name), encoding="utf-8").read()
    if "_remember" in src:
        _leaks.append(name)
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)

_carry_src = inspect.getsource(PriceHistory.carry)
_tree = ast.parse(_carry_src.strip())
_calls = [n.func.attr for n in ast.walk(_tree)
          if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)]
check(f"...and `carry()` records BEFORE it fills, so a feed carrying yesterday "
      f"alongside today refreshes the profile before consuming it {_calls}",
      _calls.index("_remember") < _calls.index("yesterday"))


# --------------------------------------------------------------------------- #
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("5. AC4 - four degraded stores, four reasons, 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)))

_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 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")

_h13 = store()
_ph.PriceHistory._remember(
    _h13, series_of(YDAY_H, YESTERDAY_MWH), NOW)
_ph.PriceHistory._prune(_h13, NOW + timedelta(days=4000))
check("pruning is by SORT ORDER, not by arithmetic against the clock: a clock "
      "that jumped four thousand days does not delete the record, which would "
      "empty the store on exactly the morning the fill is for",
      _h13.yesterday(NOW).kind == "series")


# --------------------------------------------------------------------------- #
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("...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)

# 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")