admin/goodwe-addon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
SAFETY-34
goodwe-addon/goodwe_controller/test_price_history.py
T
glenn schrooyenandClaude Opus 5.5 770988c902
checks / unit (pull_request) Successful in 8m44s
Details
TEST-26: a missing value in a check is a named FAIL, not a traceback 
Checks and prints that indexed an empty list, exhausted next(), or formatted
None killed the rest of their file (FCST-06). Read them through the new
named_absence.py (nth/first/field/given/present) and conjoin a presence term.

- every site in the brief, plus the family the new scan and a sweep found
- test_named_absence.py: scans every test_*.py for unguarded filtered-list
  indexing, next() without a default and numeric formats on maybe-None
  names; proven to bite; prints what it cannot see, counted
- test_price.py 16a: the `_T17_CAPS is _T17_CAPS` tautology now grades the
  two assemble() calls' arguments
- test_control.py: coverage table re-measured on runs graded to the end

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BwRLw7eKYj2byVdwB6kbjU
2026-09-25 00:06:58 +02:00

2093 lines
107 KiB
Python
Raw Permalink Blame History

"""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.
AC5 (FCST-05) THE CACHE'S OWN CONTRACT HOLDS FOR EVERY BYTE AND EVERY ROW.
A file holding one non-UTF-8 byte no longer stops the add-on from
starting - `_load()` caught `(json.JSONDecodeError, OSError)` and
`UnicodeDecodeError` is neither, so it escaped `PriceHistory.__init__`
and `Controller.__init__` does not guard it. And a malformed `start`
under the day the feed is WRITING to no longer kills the tariff plan
for good: the write path shares `_point_of()` with the read path,
drops the row, says so, and `_save()` rewrites the record clean.
⚠️ THE STATUS ROW IS PART OF THIS. It said `'2026-01-15 (1 h)'` about a
store whose every refresh was raising - health it did not have, in a
unit that was not its own (it counts POINTS).
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
from named_absence import first, nth
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 first((u for u in bday_hours(bdate) if brussels(u).hour == hh),
f"{bdate} {hh:02d}:00 Brussels")
# ⚠️ 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
# ⚠️ FCST-05's HIGH, AS BYTES. A cache that is structurally valid JSON up to
# the point where one byte is not UTF-8 - which is what a truncated write, a
# half-flushed page or a foreign writer produces. Written with `open(..., "wb")`
# and never through a text handle, or the fixture cannot hold the byte it is
# named for. `0xff` is not a valid UTF-8 start byte in any position.
_UTF8_BOMB = b'{"version": 1, "days": {"2026-01-14": "\xff\xfe"}}'
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):
"""A `generic_dayahead` body in the COLUMNAR shape the real feed publishes.
⚠️ TARIFF-08. This used to build `{"data": [{"start": ...,
"price_eur_mwh": ...}]}` - a shape `api.energy-charts.info` has never
published and that the shipped parser was nevertheless written to expect.
Every price fixture in the repository fed that invented shape, so the
suite graded the parser against the parser's own assumption while the live
plugin raised `SourceUnavailable` on every fetch. Real captured bodies are
replayed in test_price.py section 2b; the fixtures here are SYNTHETIC
prices in the REAL shape, because this file is grading the fill and the
history, not the parse.
"""
return json.dumps({
"unix_seconds": [int(u.timestamp()) for u in instants],
"price": [table[brussels(u).hour] for u in instants],
"unit": "EUR / MWh",
"license_info": "CC BY 4.0 (creativecommons.org/licenses/by/4.0) "
"from Bundesnetzagentur | SMARD.de",
"deprecated": False}).encode()
def transport_for(body):
async def _t(url, timeout_s):
return body
return _t
# ⚠️ NO `price_feed_url` (TARIFF-09). An override is used VERBATIM and the
# plugin appends no date range to it, so a CFG that sets one - even to the
# feed's own address - would exercise the override path in every check in this
# file and never the shipped default. The override has its own grading in
# test_price.py section 2c.
CFG = {"price_provider": "generic_dayahead",
"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("1c. TARIFF-09 - the feed is now ASKED for the horizon, and tomorrow "
"comes back FIRM")
#
# ⚠️ WHAT SECTION 1 COULD NOT SEE, AND WHY. Everything above grades the FILL
# against a feed that published one day, and it was right to: FCST-04 covers
# the window between midnight and the ~13:00-14:00 CET auction. But
# `generic_dayahead.fetch()` accepted `since`/`until` and requested the bare
# `?bzn=BE`, which answers one Brussels day AT EVERY HOUR OF EVERY DAY - so the
# one-day feed was not the pre-publication state, it was the ONLY state, and
# §8.3 dropped every hour past today from the trade set as provisional. The
# optimiser never saw tomorrow as firm.
#
# Measured on the merged tree (4257e38) at 2026-09-10T09:00+02:00 over a 48 h
# horizon, live feed, yesterday seeded: 15 firm whole hours and 33 carried,
# and BOTH ends of the permitted cycle inside 2026-09-10. At 23:00 it was 1
# firm and 47 carried, and the firm spread no longer cleared the round-trip
# cost at all. The ceiling was structural - at hour H a one-day window gives at
# most 24-H firm hours - which is why no amount of carrying could fix it.
#
# The two plans below differ in ONE thing: what the feed answered. The code
# path, the window, the store and the instant are identical, so the difference
# is the horizon and nothing else (habit 7).
TOMORROW_D = "2026-01-16"
TOMORROW_H = bday_hours(TOMORROW_D)
# ⚠️ TOMORROW'S PRICES MUST NOT OVERLAP TODAY'S OR YESTERDAY'S, or a point's
# day of origin is not recoverable from its number and every claim below is
# about a coincidence. Asserted, not assumed.
TOMORROW_MWH = {h: 300.0 + h for h in range(24)}
TOMORROW_MWH[18] = 900.0 # the dear evening TOMORROW: the global peak of
# the 48 h window, and it is on the second day
PRICE_AT = {u: TODAY_MWH[brussels(u).hour] for u in TODAY_H}
PRICE_AT.update({u: TOMORROW_MWH[brussels(u).hour] for u in TOMORROW_H})
def body_at(instants):
"""A columnar body priced PER INSTANT, so two days can differ at the same
local hour. `feed_body()` keys on the local hour alone and cannot."""
return json.dumps({
"unix_seconds": [int(u.timestamp()) for u in instants],
"price": [PRICE_AT[u] for u in instants],
"unit": "EUR / MWh",
"license_info": "CC BY 4.0 (creativecommons.org/licenses/by/4.0) "
"from Bundesnetzagentur | SMARD.de",
"deprecated": False}).encode()
def recording(body):
"""A transport that serves bytes and remembers the URL it was asked for."""
seen = []
async def _t(url, timeout_s):
seen.append(url)
return body
_t.seen = seen
return _t
_vals_today = {TODAY_MWH[h] for h in range(24)}
_vals_tmrw = {TOMORROW_MWH[h] for h in range(24)}
_vals_yday = {YESTERDAY_MWH[h] for h in range(24)}
check(f"the three days' price tables share no value at all "
f"({len(_vals_today)}/{len(_vals_tmrw)}/{len(_vals_yday)} distinct), so "
f"which day a point came from is recoverable from the point",
not (_vals_today & _vals_tmrw) and not (_vals_tmrw & _vals_yday)
and not (_vals_today & _vals_yday))
check(f"...and both fixture days are 24 Brussels hours long "
f"({len(TODAY_H)}/{len(TOMORROW_H)}), found by search rather than by "
f"adding 24 UTC hours",
len(TODAY_H) == 24 and len(TOMORROW_H) == 24)
# THE TWO FEED STATES, one code path.
_t_two = recording(body_at(TODAY_H + TOMORROW_H))
_h_two = store()
_ph.PriceHistory._remember(_h_two, series_of(YDAY_H, YESTERDAY_MWH), NOW)
_pub = run(resolve_tariff(CFG, NOW, _until, transport=_t_two, now=NOW,
history=_h_two))
_t_one = recording(body_at(TODAY_H))
_h_one = store()
_ph.PriceHistory._remember(_h_one, series_of(YDAY_H, YESTERDAY_MWH), NOW)
_pre = run(resolve_tariff(CFG, NOW, _until, transport=_t_one, now=NOW,
history=_h_one))
# ⚠️ THE REQUEST, WHICH IS THE THING THIS CARD CHANGED. Both calls ask for the
# same three Brussels days; only the answer differs. The expected dates are
# spelled out here rather than derived from the module that builds them.
check(f"the shipped plugin asks the feed for the Brussels days the horizon "
f"touches, on both calls -> {_t_two.seen}",
_t_two.seen == _t_one.seen
== ["https://api.energy-charts.info/price?bzn=BE"
"&start=2026-01-15&end=2026-01-17"])
check("...and it is no longer the bare URL that answers a single day, which "
"is what made the one-day state permanent rather than pre-publication",
"start=" in _t_two.seen[0] and "end=" in _t_two.seen[0])
def split(plan):
"""(firm, carried), and ([], []) for a plan that observed no series.
⚠️ THE EMPTY ANSWER IS DELIBERATE. A mutation that turns the short
pre-publication answer into `no_data` makes `plan.read.points` raise
`Absence` by design, and an unguarded read here would take this file down
with a traceback rather than reporting a named FAIL - "the suite is red"
and "the check that watches this is red" are different claims
(MUTATION.md, reporting a mutation, point 3). With this guard the counts
below go red BY NAME, which is what a mutation report can cite.
"""
if not plan.has_prices:
return [], []
firm = [p for p in plan.read.points if not p.provisional]
carried = [p for p in plan.read.points if p.provisional]
return firm, carried
def local_days(points):
return sorted({brussels(p.start).strftime("%Y-%m-%d") for p in points})
_f2, _c2 = split(_pub)
_f1, _c1 = split(_pre)
# The expected counts are re-derived from the fixture by set arithmetic: the
# horizon hours the feed printed, versus the horizon hours it did not.
_window = [SINCE_HOUR + timedelta(hours=n) for n in range(HORIZON_H)]
_two_pub_utc = {u.astimezone(UTC) for u in TODAY_H + TOMORROW_H}
_one_pub_utc = {u.astimezone(UTC) for u in TODAY_H}
_want_f2 = len([u for u in _window if u.astimezone(UTC) in _two_pub_utc])
_want_f1 = len([u for u in _window if u.astimezone(UTC) in _one_pub_utc])
check(f"WITH TOMORROW PUBLISHED the 48 h plan is {len(_f2)} firm hours and "
f"{len(_c2)} carried; with only today, {len(_f1)} firm and {len(_c1)} "
f"carried. Re-derived from the fixture: {_want_f2} and {_want_f1}",
len(_f2) == _want_f2 and len(_f1) == _want_f1
and len(_f2) + len(_c2) == HORIZON_H
and len(_f1) + len(_c1) == HORIZON_H
and len(_f2) > len(_f1))
check(f"...and the firm hours now span TWO Brussels local days "
f"{local_days(_f2)}, where the one-day answer spans exactly one "
f"{local_days(_f1)}. This is the acceptance criterion, and the ONE thing "
f"no amount of carry-forward could ever produce: a carried hour is "
f"provisional by construction",
len(local_days(_f2)) == 2 and len(local_days(_f1)) == 1
and local_days(_f2) == [TODAY_D, TOMORROW_D])
# ⚠️ THE TAIL IS STILL CARRIED, WHICH IS FCST-04 STILL DOING ITS JOB. A 48 h
# horizon asked for at 09:17 runs 9 hours into a THIRD Brussels day, and no
# day-ahead market publishes that far. The fix widens what is firm; it does not
# claim the market publishes more than it does.
check(f"FCST-04 still fills the remainder: the {len(_c2)} hours past the "
f"market's own horizon are carried from yesterday and flagged, on a "
f"third Brussels day {local_days(_c2)} - so the plan still covers all "
f"{HORIZON_H} hours",
len(_c2) > 0 and local_days(_c2) == ["2026-01-17"]
and all(p.provisional for p in _c2))
check("...and the plan is still a provider plan reporting itself provisional, "
"not a §11.4 fallback",
_pub.origin == "provider" and _pub.provisional)
# ⚠️ THE PRE-PUBLICATION ANSWER IS A SERIES OF WHAT EXISTS, NOT `no_data`.
# Asking for tomorrow before the auction returns fewer days than requested.
# A `no_data` there would switch a working site's arbitrage off every morning.
check(f"the short answer stays a SERIES of what exists - {_pre.read.kind}, "
f"{len(_pre.read.points)} hours, origin {_pre.origin!r} - and is never "
f"no_data and never a truncated series presented as complete "
f"({len(_c1)} of its hours are flagged provisional)",
_pre.read.kind == "series" and _pre.origin == "provider"
and len(_pre.read.points) == HORIZON_H and len(_c1) > 0)
# --- the trade set, which is where the whole thing lands ------------------- #
_A9 = dict(peak_guard_active=False, min_spread_eur_kwh=0.0,
uncertainty=frozenset())
_v2 = _pub.arbitrage(**_A9)
_v1 = _pre.arbitrage(**_A9)
check("both plans still ALLOW a trade, so the two verdicts below are compared "
"against each other and not against a refusal",
_v2.kind == "allowed" and _v1.kind == "allowed")
_v2 = _v2 if _v2.kind == "allowed" else None
_v1 = _v1 if _v1.kind == "allowed" else None
# ⚠️ NON-DEGENERACY, DERIVED. The global cheapest firm hour must be on TODAY
# and the global dearest on TOMORROW, or "the trade set spans two days" is a
# property of the arithmetic rather than of the fix.
def _day_of(v, end):
"""The Brussels day of one end of a permitting verdict, or "" for a
refusal - so a mutation that turns the verdict into a refusal is a named
FAIL below rather than an `Absence` traceback."""
return "" if v is None else brussels(getattr(v, end).start).strftime(
"%Y-%m-%d")
_lo2, _hi2 = _day_of(_v2, "cheapest"), _day_of(_v2, "dearest")
_lo1, _hi1 = _day_of(_v1, "cheapest"), _day_of(_v1, "dearest")
check(f"WITH TOMORROW FIRM the trade set contains hours from BOTH days: "
f"buy on {_lo2}, sell on {_hi2}",
sorted({_lo2, _hi2}) == [TODAY_D, TOMORROW_D])
check(f"...where the one-day answer's buy AND sell are both inside {_lo1} - "
f"a trade set that is a subset of ONE calendar day, structurally, which "
f"is the defect this card is about",
_lo1 == _hi1 == TODAY_D)
_s2 = None if _v2 is None else _v2.spread_eur_kwh
_s1 = None if _v1 is None else _v1.spread_eur_kwh
check(f"...and the cross-day spread is strictly larger ({_s2} vs {_s1} "
f"EUR/kWh), so the second day is not merely visible but PRICED",
_s2 is not None and _s1 is not None and _s2 > _s1)
check("...and both ends of the permitted cycle are FIRM, which "
"`_firm_or_die()` asserts at construction and §8.3 requires",
_v2 is not None and not _v2.cheapest.provisional
and not _v2.dearest.provisional)
# --------------------------------------------------------------------------- #
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. Belgian DAY-AHEAD settles per QUARTER
# - the 15-minute MTU took effect on 2025-10-01, so `generic_dayahead` reads 96
# points a day - and `_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 = nth(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((_qu,
100.0 + 40.0 * (_n % 4) + brussels(_qu).hour))
# ⚠️ COLUMNAR, like the real feed - see feed_body(). And 900 s spacing is not a
# hypothetical here: Belgium moved to a 15-minute MTU on 2025-10-01 and
# `api.energy-charts.info` has answered 96 points a day ever since, which is
# exactly what test_price.py section 2b replays from a captured body.
_qbody = json.dumps({
"unix_seconds": [int(_u.timestamp()) for _u, _v in _qrows],
"price": [_v for _u, _v in _qrows],
"unit": "EUR / MWh",
"license_info": "CC BY 4.0 (creativecommons.org/licenses/by/4.0) "
"from Bundesnetzagentur | SMARD.de",
"deprecated": False}).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[0] 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)))
# ⚠️ AND THE BYTE CLASS THAT FIXTURE CANNOT REACH - FCST-05's HIGH. The row
# above is INVALID JSON but it is PERFECTLY VALID UTF-8, so it is decoded and
# then refused by `json`, which raises `json.JSONDecodeError`. A file holding
# one byte that is not UTF-8 at all fails EARLIER, in the decode `open(...,
# encoding="utf-8")` performs before `json` sees a character, and
# `UnicodeDecodeError` is `ValueError -> UnicodeError -> UnicodeDecodeError`:
# NOT a `JSONDecodeError`, and NOT an `OSError` either. `_load()` caught
# `(json.JSONDecodeError, OSError)`, so it escaped `_load()` -> `_open()` ->
# `PriceHistory.__init__` - unguarded in `Controller.__init__` - AND THE ADD-ON
# DID NOT START, on a cache whose own contract says "a corrupt cache must not
# stop the controller".
#
# The precondition is asserted first, because it is the whole reason the check
# above could not see this: if `"{ this is not json"` were undecodable, the two
# fixtures would be one fixture and this would grade nothing (MUTATION.md
# habit 1).
check("the two unreadable-file fixtures are DIFFERENT FAILURES: the one above "
"is valid UTF-8 that is not JSON, and the one below is not UTF-8 at all. "
"A file that is merely bad JSON cannot reach the decode step, which is "
"why the suite had this hole",
open(_p6, "rb").read().decode("utf-8") == "{ this is not json"
and raises(lambda: _UTF8_BOMB.decode("utf-8"), UnicodeDecodeError))
_d6b = tmpdir()
_p6b = os.path.join(_d6b, "price_history.json")
open(_p6b, "wb").write(_UTF8_BOMB)
_h6b = derived(lambda: PriceHistory(_p6b))
check(f"a single non-UTF-8 byte in the cache does NOT stop the store from "
f"being constructed {_h6b!r}. `PriceHistory(...)` is unguarded in "
f"`Controller.__init__`, so an exception here is an add-on that does not "
f"start - which is what shipped",
isinstance(_h6b, PriceHistory))
check(f"...and it degrades the SAME WAY the bad-JSON file does: the reason is "
f"kept, named and reported "
f"{(_h6b.last_error if isinstance(_h6b, PriceHistory) else _h6b)!r}",
isinstance(_h6b, PriceHistory)
and "unreadable" in (_h6b.last_error or "")
and "codec can't decode byte" in (_h6b.last_error or "")
and _h6b.yesterday(NOW).kind == "no_data"
and "unreadable" in why(_h6b.yesterday(NOW)))
_w6b = derived(lambda: _h6b.words)
_plan6b = derived(lambda: run(resolve_tariff(
CFG, NOW, _until, transport=transport_for(_body_today),
now=NOW, history=_h6b)).origin)
check(f"...and NOTHING downstream raises on it either: not `words` ({_w6b!r}), "
f"not the shipped `resolve_tariff()` path ({_plan6b!r}). §6: absent "
f"machinery must not disable the loop, and a JSON file is machinery",
isinstance(_w6b, str) and _plan6b == "provider")
_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, the POINT COUNT and the "
f"HOUR COUNT {words_of(_h_words)!r} - 'a cache exists' and 'the cache "
f"holds yesterday, all 24 hours of it' are different facts, and only "
f"the second says the horizon will be full tomorrow morning",
words_of(_h_words) == f"{YDAY_D} (24 points over 24 h)")
# ⚠️ AND THE TWO NUMBERS ARE NOT THE SAME NUMBER - FCST-05's LOW. The row
# shipped as `f"{d} ({_n_points(rec)} h)"`, which counts POINTS and prints them
# under the HOUR's name. On an hourly feed the two are equal and nothing shows;
# on the quarter-hourly feed this add-on already supports (`_published_by_hour`
# keeps a TUPLE per hour precisely for it, and §7 below drives one) the row
# read '2026-01-13 (96 h)' - 96 "h" for a 24-hour day. That is the points/hours
# conflation H1 was raised for, on the row H1's own ticket shipped.
#
# MUTATION.md's rule: this check asserts a DIFFERENCE, so the fixture is run
# with the two numbers actually made different. 96 != 24, asserted first.
_h_q = store()
_QUARTERS = [u + timedelta(minutes=15 * q) for u in YDAY_H for q in range(4)]
_ph.PriceHistory._remember(
_h_q, PriceRead.series(
[PricePoint(start=brussels(u),
eur_per_kwh=allin(YESTERDAY_MWH[brussels(u).hour]
+ brussels(u).minute),
provisional=False) for u in _QUARTERS],
PriceProduct.DAY_AHEAD, "generic_dayahead", NOW),
NOW)
_qw = words_of(_h_q)
check(f"the quarter-hourly fixture really is sub-hourly and its two counts "
f"really differ ({len(_QUARTERS)} points, "
f"{len({brussels(u).hour for u in _QUARTERS})} local hours), or 'points' "
f"and 'hours' are the same observation and the row grades nothing",
len(_QUARTERS) == 96
and len({brussels(u).hour for u in _QUARTERS}) == 24
and len({u.minute for u in _QUARTERS}) == 4)
check(f"...and the row reports BOTH, each under its own unit: {_qw!r}. "
f"A day of 96 quarter-hourly prints is 96 points over 24 h, and the "
f"figure that says whether tomorrow's fill has a whole day to carry is "
f"the HOUR one",
_qw == f"{YDAY_D} (96 points over 24 h)")
check(f"...and specifically the row does NOT say '96 h' {_qw!r} - a point "
f"count wearing the hour's name is H1's defect, and this row is where "
f"an operator would have read it",
isinstance(_qw, str) and "(96 h)" not in _qw and " (96 points " in _qw)
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, _h6b, _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),
("a non-UTF-8 byte in the file", _h6b),
("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("5c. FCST-05 - the WRITE path is as defensive as the read path")
# THE DEFECT. `_remember()` merged a day's held points with
#
# held = {q["start"]: q for q in raw
# if isinstance(q, dict) and "start" in q}
# rec["points"] = sorted(held.values(), key=lambda q: q["start"])
#
# which trusts a value read off /data to be a HASHABLE, SORTABLE STRING. It is
# whatever is in the file. `_usable_days()` required `rec["points"]` to be a
# LIST and never looked at the elements, and `_parse_points()` - the READ path -
# was properly defensive, so `yesterday()` survived and the §5 `_p8` fixture
# passed. Two standards for what a stored point is; the weaker one on the path
# that WRITES.
#
# ⚠️ AND THE PLACEMENT IS THE WHOLE FIXTURE. `_p8` files its malformed rows
# under a day the feed is NOT writing to, so `_remember()` never touches that
# record and the defect is invisible. What reaches it is a malformed row under
# TODAY - the record the running feed merges into on every refresh - so that is
# asserted before anything is graded (MUTATION.md habit 1), and the harmless
# placement is run as the accepted baseline it varies from (habit 7).
#
# ⚠️ AND THE BLAST RADIUS IS WHY IT IS NOT MERELY UNTIDY. `carry()` is reached
# from `resolve_tariff()` from `run_tariff_refresh`, which catches `Exception`:
# the add-on stays UP and EVERY refresh raises, forever. The plan retires after
# TARIFF_MAX_AGE_S and arbitrage and the cheap window are off permanently, with
# no self-heal, because `_remember()` raises before it reaches `_save()`.
_BAD_STARTS = (
("an int", 7), # TypeError: '<' between 'str' and 'int'
("a dict", {"a": 1}), # TypeError: unhashable as a dict key
("null", None), # TypeError: '<' between 'str', 'NoneType'
("a list", [1, 2]), # TypeError: unhashable as a dict key
("a string that is not an instant", "nope"), # sorts fine, never a point
)
def _seeded(day, bad_start):
"""A store on disk holding ONE malformed row under `day`, plus a good one."""
d = tmpdir()
path = os.path.join(d, "price_history.json")
good = {"start": brussels(at_local(day, 5)).isoformat(),
"eur_per_kwh": allin(YESTERDAY_MWH[5]), "provisional": False}
json.dump({"version": SCHEMA, "days": {day: {
"product": "dynamic_dayahead", "provider": "generic_dayahead",
"points": [good, {"start": bad_start, "eur_per_kwh": 0.2,
"provisional": False}]}}},
open(path, "w", encoding="utf-8"))
return path, PriceHistory(path)
_TODAY_FEED = series_of(TODAY_H, TODAY_MWH)
check(f"the fixture files its malformed row under {TODAY_D}, which is exactly "
f"the Brussels day the feed being carried writes to "
f"{sorted({_ph._bdate(p.start) for p in _TODAY_FEED.points})} - under any "
f"other day `_remember()` never opens the record and this whole section "
f"grades nothing, which is why the suite already had this hole",
{_ph._bdate(p.start) for p in _TODAY_FEED.points} == {TODAY_D})
# The accepted baseline: the SAME row, one day over. Shipped and broken alike
# pass this, and that is the point of running it (habit 7).
_pB, _hB = _seeded(YDAY_D, 7)
_cB = derived(lambda: _hB.carry(_TODAY_FEED, since=NOW, until=_until, stamp=NOW))
check("baseline: the same malformed row under a day the feed is NOT writing "
"to is carried without incident - one variable moves between this and "
"every check below, and it is the DAY",
not isinstance(_cB, _Raised) and _cB.kind == "series")
for _shape, _bad in _BAD_STARTS:
_pM, _hM = _seeded(TODAY_D, _bad)
_wload = words_of(_hM)
_c1 = derived(lambda h=_hM: h.carry(_TODAY_FEED, since=NOW, until=_until,
stamp=NOW))
_saw = repr(_c1) if isinstance(_c1, _Raised) else _c1.kind
check(f"a `start` that is {_shape} ({_bad!r}) under TODAY's own record "
f"does not take the refresh down: `carry()` returns a series "
f"({_saw}). It raised "
f"`TypeError` out of the shipped path, and `run_tariff_refresh` "
f"catches `Exception`, so the add-on stayed up while every refresh "
f"failed for good",
not isinstance(_c1, _Raised) and _c1.kind == "series")
# ⚠️ THE VERDICT IS NOT THE CLAIM. "It did not raise" is satisfied by a
# store that silently kept the poison row; what this section promises is
# that the row is GONE FROM THE FILE, so the next start is clean.
_disk = json.load(open(_pM, encoding="utf-8"))
_rows = _disk["days"][TODAY_D]["points"]
# DERIVED, not a constant: the surviving rows are exactly the DISTINCT
# instants the feed published for that day plus the fixture's one good
# seeded row - which is 05:00, an hour the feed also prints, so it merges.
_want = len({p.start.astimezone(UTC) for p in _TODAY_FEED.points}
| {at_local(TODAY_D, 5).astimezone(UTC)})
check(f"...and the row is DROPPED FROM /data, not carried: {len(_rows)} "
f"row(s) on disk against {_want} distinct instants, every one of "
f"them a point. `_remember()` used to raise BEFORE `_save()`, so "
f"there was no self-heal at all - the store came back poisoned on "
f"every restart",
all(_ph._point_of(r) is not None for r in _rows)
and len(_rows) == _want)
# ...and the store stops reporting health, at BOTH moments it can be read:
# when it is opened, and after the refresh that cleaned it.
# ⚠️ THE COUNTS ARE ASSERTED TOO, NOT ONLY THE NOTE. The seeded record holds
# ONE point and one row that is not one, so a row printing "2 points" would
# be reporting the poison as held prices - which is the half of this defect
# that reached the operator: `words` said "2026-01-15 (1 h)" while every
# tariff refresh was dying on that record.
check(f"...and the status row does not call that store healthy - at open "
f"{_wload!r}: ONE point held, and the row that is not a point named "
f"rather than counted",
isinstance(_wload, str)
and _wload.startswith(f"{TODAY_D} (1 point over 1 h) [")
and "1 malformed point row(s) in" in _wload)
# ⚠️ AND THE TWO NOTES ARE GRADED APART, BECAUSE THEY MASK EACH OTHER
# OTHERWISE (MUTATION.md TEST-16). `_usable_days()` records what it FOUND
# at open and `_remember()` records what it DROPPED on the refresh; both
# land in `last_error`, so a check asking only "does the row mention a
# malformed point" is answered by whichever rung is still standing - the
# first draft of this check was exactly that, and deleting `_remember()`'s
# note left the whole 29-file suite green at -1 B. The tense is the
# difference an operator reads: "in" is still there, "dropped from" is
# gone, and after the refresh only the second is true.
_wafter = words_of(_hM)
check(f"...nor after the refresh that dropped it {_wafter!r}. It said "
f"'{TODAY_D} (1 h)' - the status row reporting a full cache while "
f"every tariff refresh was dying on it",
isinstance(_wafter, str)
and "malformed point row(s) dropped from" in _wafter
and TODAY_D in _wafter)
check(f"...and the row MOVED from what was found to what was dropped "
f"{_wload.split('[')[-1]!r} -> {_wafter.split('[')[-1]!r}, so the "
f"two notes are not one note read twice",
isinstance(_wload, str) and isinstance(_wafter, str)
and "malformed point row(s) in" in _wload
and "malformed point row(s) in" not in _wafter
and "malformed point row(s) dropped from" not in _wload)
# ...and a SECOND refresh over the same store is clean, which is the state
# a restarted add-on comes up in.
_hM2 = PriceHistory(_pM)
check(f"...and the store reopened from that file is CLEAN: the whole day "
f"held and nothing to report {words_of(_hM2)!r}, so the note clears "
f"itself rather than becoming permanent furniture",
words_of(_hM2) == f"{TODAY_D} ({_want} points over {_want} h)"
and _hM2.last_error is None)
# ...and the shipped path, end to end, through the caller that catches
# `Exception` and would otherwise have hidden this for good.
_pP, _hP = _seeded(TODAY_D, 7)
_o1 = derived(lambda: run(resolve_tariff(
CFG, NOW, _until, transport=transport_for(_body_today), now=NOW,
history=_hP)).origin)
_o2 = derived(lambda: run(resolve_tariff(
CFG, NOW, _until, transport=transport_for(_body_today), now=NOW,
history=_hP)).origin)
check(f"the SHIPPED `resolve_tariff()` path survives it TWICE ({_o1!r}, "
f"{_o2!r}) - once for the refresh that finds the poison and once for "
f"the one after it. A guard that refuses once and dies on the next call "
f"is a delay, not a guard (MUTATION.md habit 9)",
_o1 == "provider" and _o2 == "provider")
# ⚠️ AND THE MERGE KEY IS THE INSTANT, WHICH IS WHAT RULE 4 SAYS. The old key
# was `q["start"]` - the SPELLING - so the same instant written at two offsets
# was held twice, and the lexicographic sort beside it put the second 02:00 of
# the 25-hour October day (`+01:00`) BEFORE the first (`+02:00`). Aware
# datetimes hash and compare by instant, so the validated `start` is the key.
_d15 = tmpdir()
_p15 = os.path.join(_d15, "price_history.json")
_alt = at_local(YDAY_D, 6)
json.dump({"version": SCHEMA, "days": {YDAY_D: {
"product": "dynamic_dayahead", "provider": "generic_dayahead",
"points": [{"start": _alt.astimezone(UTC).isoformat(),
"eur_per_kwh": 0.999, "provisional": False}]}}},
open(_p15, "w", encoding="utf-8"))
_h15 = PriceHistory(_p15)
check(f"the two spellings really are one instant and really are two strings "
f"({_alt.astimezone(UTC).isoformat()!r} vs "
f"{brussels(_alt).isoformat()!r}), or 'merged by instant' and 'merged by "
f"spelling' are the same observation",
_alt.astimezone(UTC).isoformat() != brussels(_alt).isoformat()
and _alt.astimezone(UTC) == brussels(_alt))
_ph.PriceHistory._remember(
_h15, series_of([_alt], YESTERDAY_MWH), NOW)
_y15 = _h15.yesterday(NOW)
_n15 = len(_y15.points) if _y15.kind == "series" else _y15.kind
_v15 = _y15.points[0].eur_per_kwh if _y15.kind == "series" else None
check(f"...and a revision of that hour REPLACES it rather than sitting beside "
f"it: {_n15} point held, priced {_v15} and not 0.999. Rule 4 says "
f"'MERGE BY INSTANT'; the key was the string",
_y15.kind == "series" and len(_y15.points) == 1
and abs(_y15.points[0].eur_per_kwh - allin(YESTERDAY_MWH[6])) < 1e-9)
# --------------------------------------------------------------------------- #
print()
print("5d. FCST-06 - the OTHER way to be a store that cannot persist")
# THE DEFECT. Section 5 above grades "a store that cannot persist records
# nothing" against `not ok` - the missing-durable-volume SYMPTOM - and that is
# the only condition the rule was ever enforced on. `ok` stays TRUE on a volume
# that is present and cannot be written: a read-only remount, a full disk. In
# that state `_remember()` had already populated `self._days`, `_save()` logged
# and returned, and the store went on answering a series for the rest of the
# process's life. Measured on 5059e28, twice, with a `/data` bind-mounted
# read-only and again on a full tmpfs:
#
# last_error = 'could not write ([Errno 30] Read-only file system)'
# words = '2026-01-14 (24 points over 24 h); 2026-01-15 (24 points
# over 24 h) [could not write (...)]'
# yesterday()= series, 24 points
# ON DISK = {} <- and gone on the next restart
#
# `words` and `yesterday()` were describing two different stores, which is
# exactly what section 5's own comment refuses - reached by the condition the
# rule did not name.
#
# ⚠️ THE LEVERS ARE REAL FAILURES, NOT PATCHED ONES, AND THERE ARE TWO OF THEM
# ON PURPOSE. The fix is written against the OUTCOME - `_save()` answering "no"
# - and not against an errno, so one cause would grade a coincidence. A
# DIRECTORY where the file goes fails `os.replace`; a volume that moves out
# from under an already-open store fails `mkstemp`. Different syscall,
# different errno, one answer (habit 10: grade the family).
def npts(read):
"""A read's point count, or "no_data". Never raises.
⚠️ NOT `len(read.points)`. On a `no_data` that attribute RAISES `Absence`
by design, so a mutation that empties a store where this section expects
one would take the harness down with a traceback instead of reporting a
FAIL - and MUTATION.md's rule is that "it answered nothing" and "it broke
on the way to answering" must not grade the same. Found by driving M4.
"""
return read.match(series=lambda r: len(r.points),
no_data=lambda _r: "no_data")
_d16 = tmpdir()
_p16 = os.path.join(_d16, "price_history.json")
_h16 = PriceHistory(_p16)
os.mkdir(_p16) # `os.replace` cannot complete onto a directory
check(f"the fixture is an `ok` store on a directory that IS there "
f"(ok={_h16.ok}, last_error={_h16.last_error!r}) - not the "
f"no-durable-volume case section 5 already grades, or this section is "
f"that one again (habit 1)",
_h16.ok and _h16.last_error is None and os.path.isdir(_d16))
_carried16 = _h16.carry(series_of(YDAY_H + TODAY_H, TODAY_MWH),
since=NOW, until=_until, stamp=NOW)
_y16 = _h16.yesterday(NOW)
check(f"a write that could not complete is REPORTED and leaves the store "
f"`ok` (ok={_h16.ok}, {(_h16.last_error or '')[:44]!r}). `ok` is about "
f"the DIRECTORY, and a full volume and a read-only remount both leave "
f"it True - which is why testing `ok` alone could not see this",
_h16.ok and "could not write" in (_h16.last_error or ""))
check(f"...and the file really does not hold the day "
f"(isfile={os.path.isfile(_p16)}), or 'it answers what is not there' is "
f"not a measurement (habit 1)",
not os.path.isfile(_p16))
check(f"A STORE THAT COULD NOT WRITE RECORDS NOTHING, exactly as one with no "
f"durable volume does: it holds {sorted(_h16._days)} and `yesterday()` "
f"is {_y16.kind!r}. Retaining the merge is an in-memory yesterday that "
f"answers all day and is gone on the next restart - the "
f"silently-forgetting cache this module's docstring exists to refuse",
_h16._days == {} and _y16.kind == "no_data")
check(f"...and `words` and `yesterday()` AGREE about that store "
f"{_h16.words[:44]!r}: neither names a day the file does not hold, and "
f"both name the write failure. They disagreed - `words` printed two "
f"full days BESIDE the error while `yesterday()` answered 24 points",
YDAY_D not in _h16.words and TODAY_D not in _h16.words
and "could not write" in _h16.words
and _y16.kind == "no_data" and "could not write" in why(_y16))
check(f"...and it DEGRADES rather than raising: the carry still returned a "
f"real series ({_carried16.kind!r}, {npts(_carried16)} points) with "
f"nothing carried. §6 and this module's opening - an add-on that will "
f"not start because /data filled up is a WORSE defect than the one this "
f"section is about",
_carried16.kind == "series"
and not _carried16.match(
series=lambda r: any(p.provisional for p in r.points),
no_data=lambda _r: True))
_plan16 = derived(lambda: run(resolve_tariff(
CFG, NOW, _until, transport=transport_for(_body_today), now=NOW,
history=_h16)).origin)
check(f"...and the SHIPPED `resolve_tariff()` path is unmoved by it "
f"({_plan16!r}): a provider plan, no §11.4 degradation, on a store that "
f"cannot write a byte",
_plan16 == "provider")
# Habit 9: the same wrong input a SECOND time. A refresh runs 48 times a day,
# and a volume that is full at 14:00 is still full at 14:30.
_h16.carry(series_of(YDAY_H, YESTERDAY_MWH), since=NOW, until=_until,
stamp=NOW)
check(f"...and the SECOND failing refresh behaves the same rather than "
f"accumulating: still {sorted(_h16._days)} held and "
f"{_h16.yesterday(NOW).kind!r}. The write fails 48 times a day, so the "
f"call that matters is the one after the first (habit 9)",
_h16._days == {} and _h16.yesterday(NOW).kind == "no_data")
# ⚠️ AND WHAT IS KEPT IS THE LAST STATE THAT IS ON DISK, NOT AN EMPTY ONE. The
# MERGE is undone, not the store. A site whose volume fills at 14:00 goes on
# answering the yesterday it wrote at 13:30 - that record survives a restart,
# so answering it is not a promise the store cannot keep. What it stops doing
# is adding to it. This is also the fixture that separates a deep snapshot from
# `dict(self._days)`: a one-level copy shares the day RECORD the merge rebinds
# `points` on, so the "restored" store would answer the 24 points the merge
# made and the file never took.
_d17 = os.path.join(tmpdir(), "data")
os.mkdir(_d17)
_p17 = os.path.join(_d17, "price_history.json")
_h17 = PriceHistory(_p17)
_ph.PriceHistory._remember(_h17, series_of(YDAY_H[:12], YESTERDAY_MWH), NOW)
_first17 = _h17.yesterday(NOW)
check(f"a HALF day is persisted first ({npts(_first17)} points, "
f"last_error={_h17.last_error!r}), or 'the merge is undone and the "
f"store is not' grades nothing (habit 1)",
_h17.last_error is None and os.path.isfile(_p17)
and npts(_first17) == 12)
_persisted17 = open(_p17, encoding="utf-8").read()
_moved17 = _d17 + ".moved"
os.rename(_d17, _moved17) # the volume goes out from under an OPEN store:
# `mkstemp` fails, a different call and a
# different errno from the fixture above, and
# the bytes already written stay intact
_ph.PriceHistory._remember(_h17, series_of(YDAY_H[12:], YESTERDAY_MWH), NOW)
_y17 = _h17.yesterday(NOW)
_n17 = npts(_y17)
check(f"the second lever failed the write by a DIFFERENT route "
f"{(_h17.last_error or '')[:44]!r} - `mkstemp`, not `os.replace`, and "
f"ENOENT, not EISDIR. The rule is enforced on the outcome, so it must "
f"be driven by more than one cause (habit 10)",
"could not write" in (_h17.last_error or "")
and "No such file" in (_h17.last_error or ""))
check(f"...and the store keeps the last state that IS ON DISK, not an empty "
f"one: {_n17} points held - the 12 it wrote before the volume went, not "
f"the 24 the unwritten merge made, and not nothing. The MERGE is undone; "
f"the store is not",
npts(_y17) == 12)
_ph.PriceHistory._remember(_h17, series_of(YDAY_H[12:], YESTERDAY_MWH), NOW)
_y17b = _h17.yesterday(NOW)
check(f"...and again on the second failing refresh ({npts(_y17b)} points), "
f"because a volume that has gone is still gone 30 minutes later "
f"(habit 9)",
npts(_y17b) == 12)
os.rename(_moved17, _d17) # the volume comes back: what a RESTART finds
_r17 = PriceHistory(_p17).yesterday(NOW)
check(f"...and that is EXACTLY what a restart finds ({npts(_r17)} points, the "
f"file byte-for-byte as it was left), which is the whole claim: the "
f"store answers what survives a restart and nothing else",
open(_p17, encoding="utf-8").read() == _persisted17
and npts(_r17) == 12
and _r17.match(series=lambda a: _y17.match(
series=lambda b: [p.start for p in a.points]
== [p.start for p in b.points],
no_data=lambda _r: False), no_data=lambda _r: False))
# ⚠️ AND "CANNOT PERSIST" IS ONE PREDICATE, ASKED ONCE. FCST-04's M4 and
# FCST-05's HIGH were both a rule enforced on the conditions somebody thought
# of rather than on the family; this is the third of the shape and the last
# place it can be spelt twice. `_save()` returns whether the file now holds
# what memory holds - no volume is one False, an `OSError` is another - and
# `_remember()` keeps only what that answer says was kept.
check(f"`_save()` answers for BOTH symptoms and is the only thing that does: "
f"no durable volume -> {_ph.PriceHistory._save(_missing)!r}, a write "
f"that fails -> {_ph.PriceHistory._save(_h16)!r}, a store that "
f"persisted -> {_ph.PriceHistory._save(_hok)!r}",
_ph.PriceHistory._save(_missing) is False
and _ph.PriceHistory._save(_h16) is False
and _ph.PriceHistory._save(_hok) is True)
check(f"...and asking it once leaves the two symptoms with ONE behaviour: "
f"no volume holds {sorted(_missing._days)}/"
f"{_missing.yesterday(NOW).kind!r} and a failed write holds "
f"{sorted(_h16._days)}/{_h16.yesterday(NOW).kind!r}. A third cause "
f"nobody has thought of needs no third branch - which is how this defect "
f"arrived, twice",
(_missing._days, _missing.yesterday(NOW).kind)
== (_h16._days, _h16.yesterday(NOW).kind))
check("...and `_remember()` no longer tests a symptom of its own: the source "
"holds no `self.ok` test between the docstring and the merge. This one "
"IS about the code's shape - two places testing two symptoms is the "
"defect, not a behaviour (habit 4's stated exception)",
"if not self.ok:" not in inspect.getsource(_ph.PriceHistory._remember)
and "if not self.ok:" in inspect.getsource(_ph.PriceHistory._save))
# --------------------------------------------------------------------------- #
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 = first((n for n in ast.walk(_main_tree)
if isinstance(n, ast.AsyncFunctionDef)
and n.name == "run_tariff_refresh"), "run_tariff_refresh in main.py")
_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 = first((n for n in ast.walk(_main_tree)
if isinstance(n, ast.FunctionDef)
and n.name == "status_payload"), "status_payload in main.py")
_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")
Reference in New Issue View Git Blame Copy Permalink