Merge release/1.0 (OPT-04) into SAFETY-24, and fix one grader the merge broke

OPT-04 landed after the previous rebase. Conflicts were CHANGELOG.md and one
IMPORT LINE in validate.py; both refusal blocks and both option sets survive
untouched, and DOCS.md, main.py and config.yaml auto-merged.

⚠️ AND THE MERGE BROKE A CHECK THAT NEITHER CARD COULD HAVE SEEN ALONE.
test_peakguard.py graded "the first `.decide(` call in main.py" and asserted
there was exactly one. That was true when SAFETY-24 was written; OPT-04 added
`arbitrage.decide(...)`, so the merged tree had two and the assertion failed.
It failed loudly rather than grading the wrong call - the guard did its job -
but it cannot ship red.

Fixed by selecting the call by QUALIFIED NAME (`peakguard.decide`) instead of
by position, and requiring exactly one of THAT. Other modules may now call
their own decide(); this grader can no longer be moved by one of them. Verified
still discriminating: a decoy `peakguard.decide` added to main.py turns it red
on 7 assertions, and the `maintenance_soc_floor` swap still turns it red on 6.

Also folded in the review's wording finding. An assertion read "a CONTROLLER
whose SoC entity does not answer hands decide() None" while driving no
Controller. The wiring property is stronger than the one claimed, and is now
asserted on the SERVICE CALL: a Controller with no SoC never reaches the guard
at all, because `inputs_ok` is False and cycle() stops at the safety claim
first - so a `float(soc or 100)` defaulting line above the call site cannot
disarm the floor. The original decide()-level check is kept, reworded to say
what it actually grades.

Full suite green on the merged tree: 15 files.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015VXjA3LWccN1rJsk3okyPC

goodwe_controller/CHANGELOG.md

@@ -2,6 +2,99 @@
 
 ## Unreleased
 
+**The capacity-peak guard's forced discharge now has an EMS-side energy floor
+(SAFETY-24).** SAFETY-07 can command a discharge to hold a billed quarter down,
+and nothing below it consulted the state of charge — `soc_floor` belongs to the
+maintenance cycle, and freeze.py already recorded that for the grid-following
+loop "the BMS is the sole energy backstop". On the 2026-08-20 replay the guard
+asks for **878 Wh in a day**. Relying on the BMS is not absurd — it is a real
+floor and it works — but it is the pack's protection rather than the operator's,
+and a month of aggressive quarters leaves the battery repeatedly at that cut-out
+with no EMS-side record of why.
+
+`decide()` now takes `soc` and `soc_floor`, both **required** and neither
+defaulted. Above the floor nothing changes. At or below it the guard files a
+**0 W** floor instead of a sized one — one-directional, so charging stays blocked
+and no discharge is capped — and reports `floor_limited` with the reason it
+bound. It is not a trip: every stop in §11.2's table still outranks the claim,
+and the floor can never be negative, so §9.1's "discharge and export are never
+capped by the peak guard" survives as a property of the object (swept: 0 negative
+floors across 1296 combinations of load, elapsed, battery, SoC and floor).
+
+**Failing to shave is not silent.** A floor-limited cycle raises one notification
+per episode (`goodwe_peak_soc_floor`), distinct from the saturation alarm because
+the two ask the operator for opposite things, and renders its own status row.
+An **unreadable SoC is floor-limited, not a full battery** — a full battery is
+exactly the value that would disarm the floor, and this project has shipped that
+defaulting once before (`| float(100)` in the stage-5 loop).
+
+**Verified by replay, not by argument.** The 2026-08-20 capture is driven against
+a modelled pack sized from the day's own measured appetite, so it genuinely
+exhausts: without the floor it reaches the BMS cut-out and sits there for 1227
+guard cycles; with it the pack stops at 20.0 % and never touches the BMS. The
+energy withheld equals the energy between the two floors to within one cycle at
+the rail. The honest half is reported too — the worst quarter goes from 4338 W to
+5030 W, because a floor-limited guard shaves less on purpose.
+
+**Two new options, mirrored once.** `peak_soc_floor` (20) and `bms_soc_floor`
+(10), whose defaults live in `peakguard.PEAK_OPTION_DEFAULTS` — TARIFF-02's
+pattern — imported by `validate.py` and `main.py` rather than typed as literals,
+and pinned to `config.yaml` by a check that reads the file. CFG-03 still tracks
+three places. `validate.py` rule 8 refuses `peak_soc_floor <= bms_soc_floor` at
+startup, in CFG-01/MAINT-04's manner. `maintenance_soc_floor` is deliberately
+**not** reused and deliberately **not** renamed: it is a drain target, not a
+spend limit, and the rename is recorded as a decision to revisit if a third
+floor ever appears.
+
+**Price arbitrage - the first strategy that trades on a price (OPT-04).**
+Four cards built the price stack and nothing traded on it. `app/arbitrage.py`
+is §10.1's priority 3: charge in the cheapest published hour, discharge in the
+dearest, **and only when the spread covers `cycle_cost_threshold_eur`**. It is
+off by default (`arbitrage_enabled: false`) - it is the first strategy in this
+add-on that commands the battery against the meter on purpose.
+
+**It consumes the four cards rather than re-deciding anything.** Whether a
+trade is permitted at all, on which two hours, and against which widened
+threshold is `TariffPlan.arbitrage()`'s answer (TARIFF-03), and that answer
+cannot be obtained without first stating whether the peak guard is holding. So
+§9.1's "give peak shaving priority over arbitrage" is honoured
+**structurally**: there is no code path to a tradeable spread that skips the
+question. And because the claim is filed at `P_ARBITRAGE` (30) while peakguard
+files its floor at `P_PEAK` (50), the arbiter clamps an arbitrage charge even
+if that interlock were bypassed. Two independent mechanisms, neither
+re-implemented here.
+
+**A guard that never ran is not a guard that is clear.** On `meter_source: off`
+there is no telegram stream, so the peak guard is never evaluated. That third
+state is carried as `None` and it **refuses**: charging at the rail creates the
+billed quantity while nothing measures the quarter it lands in. `validate.py`
+refuses that configuration at startup too, along with arbitrage armed on a site
+with no price source at all, and a SoC floor at or above the ceiling.
+
+**§17: never export into a negative price.** The guard is on the hour being
+sold into, not on the spread - a day priced -0.30 ... -0.05 has a 0.25 EUR/kWh
+spread that clears any shipped threshold and a dearest hour that is still
+negative. Buying in a negative hour stays allowed; being paid to charge is the
+right trade.
+
+**Prices are fetched in the background, never on the tick.** `resolve_tariff()`
+may open an HTTPS connection, and the control tick is what feeds the ESP32
+heartbeat - a supplier's TLS timeout there is a supplier able to push the
+add-on towards layer 1's 30 s veto. A separate task refreshes every 30 minutes;
+the tick reads an attribute and an age bound, and a plan whose refresh has
+stopped answering retires rather than standing as if it were current.
+
+**What is honestly not measured.** The acceptance criterion asked for
+net-positivity across 12 months of *replayed* prices. This repository holds no
+price history - `scenarios/` is P1 meter data - so that criterion is **not met,
+and is not restated**. What `test_arbitrage.py` section 7 does instead is
+replay a synthetic year and assert the property that does not need real prices:
+every executed round trip cleared its own widened requirement, on all of them,
+across a year containing 522 hours priced below zero and zero discharges into
+one. And at the shipped 0.10 EUR/kWh threshold a realistic Belgian day/night
+site **does not trade at all** - reported, not tuned. That number, and the
+widening factors beside it, are one product decision and it is open.
+
 **Provisional prices, and what may not rest on one (FCST-03).**
 Day-ahead prices arrive once a day, around 14–16 h CET. Until they do, §8.3 says
 the unknown hours use yesterday's profile and are flagged `provisional: true` —

goodwe_controller/DOCS.md

@@ -420,7 +420,7 @@ help - see "Why the tuning is what it is".
 | `maintenance_start_hour` | 10 | Hour of day a due cycle begins |
 | `maintenance_discharge_w` | 2000 | Drain rate (exports the surplus). Above `max_w` it is **clamped** to `max_w` — the drain phase SETs the setpoint and the arbiter bounds it, saying so on the status page (`limited by supervised(max_w)`). Warned about, not refused: the cycle still completes, just slower |
 | `maintenance_charge_w` | 2000 | Charge ceiling, capped again by peak headroom. Must stay **at or below `max_w`** or the cheap-window shaping (`hi = -maintenance_charge_w`) contradicts the supervised limit (`lo = -max_w`), the arbiter resolves it to **0 W**, and maintenance silently never charges in a cheap window — **the add-on refuses to start** |
-| `maintenance_soc_floor` | 11 | Drain target — stay just above the inverter's own floor |
+| `maintenance_soc_floor` | 11 | Drain target — stay just above the inverter's own floor. ⚠️ **This is not `peak_soc_floor`.** It is the depth the monthly recalibration deliberately goes to, once, on purpose; `peak_soc_floor` is the depth a money-saving discharge may never reach on any day. The right values differ (11 vs 20 as shipped), which is why they are two keys |
 | `maintenance_soc_target` | 99 | Charge target |
 | `maintenance_hold_min` | 120 | Hold at full so the BMS can balance |
 
@@ -430,6 +430,8 @@ help - see "Why the tuning is what it is".
 |---|---|
 | `peak_forecast_entity` | Quarter-hour demand forecast, for capacity-tariff markets. Empty = no cap |
 | `peak_cap_w` | The site's capacity-tariff target |
+| `peak_soc_floor` (20) | **The percent below which the capacity-peak guard will not discharge the battery.** Not optional and not the same thing as `maintenance_soc_floor` — see below |
+| `bms_soc_floor` (10) | The pack's **own** cut-out percent, the number the BMS enforces whatever you set above. `peak_soc_floor` must be **strictly above** it or the add-on refuses to start |
 | `price_now_entity`, `price_avg_entity` | Dynamic tariff. Empty = never force a paid grid top-up |
 
 On a capacity-tariff site the maintenance charge is capped by the headroom left
@@ -530,11 +532,46 @@ projected over — raises one notification (`goodwe_peak_saturated`), because th
 is the one state where the money will be spent and shedding load is the only
 thing left that changes the bill.
 
-**No new option.** The ceiling is `peak_cap_w`, which already shipped. The
+**No new option for the CEILING.** It is `peak_cap_w`, which already shipped. The
 coverage threshold is `meter_max_age_s`, the freeze predicate is
 `meter_freeze_s`, and the achievable discharge comes from `max_w` and the device
 rating — every one of them a key that already exists.
 
+#### The energy floor under the forced discharge (`peak_soc_floor`)
+
+The guard above can **command a discharge**, and until SAFETY-24 nothing in the
+add-on bounded how far down it took the pack — the BMS was the only backstop.
+On the 2026-08-20 replay the guard asks for **878 Wh in a day**.
+
+That is not absurd: the BMS floor is real and it works. But it is *the pack's*
+protection, not *yours*. A month of aggressive quarters can leave the battery
+repeatedly parked at that cut-out with no record on this side of why, and the
+first symptom — a battery that is never available when you want it — looks
+exactly like the SoC drift the monthly maintenance cycle exists to correct.
+
+- **`peak_soc_floor` (20 %)** is where the forced discharge stops. Below it the
+  guard files a **0 W** floor instead of a sized one: charging stays blocked
+  (that can only ever lower your offtake), and no further discharge is asked for.
+- **It does not stop the ordinary self-consumption loop.** That loop still has no
+  SoC guard of its own. This bounds what the *peak guard* spends for money.
+- **When it binds you are told**, once per episode, with notification
+  `goodwe_peak_soc_floor` and a status-page row reading `PEAK GUARD
+  FLOOR-LIMITED`: the quarter **will** go over the ceiling, so shedding load is
+  the only thing left that changes the bill. Failing to shave is never silent —
+  the euro figure is the whole point of the guard.
+- **An unreadable SoC is treated as floor-limited, not as a full battery.** A
+  full battery is exactly the value that would disarm this floor.
+- **`bms_soc_floor` (10 %)** is the pack's own cut-out — register `0x0560` on the
+  reference GW5000S-BP, consistent with DoD 90. It exists so the add-on can
+  **refuse to start** when `peak_soc_floor` is at or below it: such a floor can
+  never be the thing that stops the discharge, and a guard that advertises a
+  backstop it does not have is the one thing this project will not ship. Equal is
+  refused too — equal buys zero margin.
+
+Why 20 %: the reference pack is 13.8 kWh usable, so 10 points of margin over the
+BMS cut-out is about 1.6× the guard's worst measured day. If your pack is
+smaller, or your quarters are worse, raise it.
+
 ### The winding-loop alarm (the output clamp)
 
 Every setpoint this add-on writes is clamped to `max_w` before it leaves the
@@ -644,15 +681,16 @@ floor of 60 s that figure is 60 W and only the margin is left.
 
 ### Energy price providers
 
-> **No *fetched* price moves energy in 0.5.x.** This release ships the
-> *plumbing*: the five provider plugins, the canonical all-in format they emit,
-> and the config that selects one. Turning a provider on fetches, validates and
-> publishes prices and changes no charging decision; the arbitrage strategy that
-> would use it is a later ticket.
+> **A fetched price moves energy only when you switch arbitrage on.** This
+> release ships the five provider plugins, the canonical all-in format they
+> emit, the config that selects one, and — new in OPT-04 — a price-arbitrage
+> strategy that trades on them. With `arbitrage_enabled: false`, which is the
+> shipped default, turning a provider on still fetches, validates and publishes
+> prices and changes no charging decision.
 >
-> The one exception is `static_profile` below. A static day/night tariff needs
-> no network, so its cheap window is computed on every cycle and does reach the
-> maintenance cycle's grid top-up.
+> `static_profile` below needs no network at all, so its cheap window is
+> computed on every cycle and reaches the maintenance cycle's grid top-up
+> whether or not arbitrage is armed.
 
 | option | meaning |
 |---|---|
@@ -804,6 +842,50 @@ to, and the site simply has no tariff until the feed returns. It is not required
 — the add-on warns rather than refusing — but filling it in is what makes that
 fallback work.
 
+### Price arbitrage (`arbitrage_enabled`)
+
+§10.1 priority 3: charge in the cheapest published hour, discharge in the
+dearest, **and only when the spread covers what the round trip costs.** Off by
+default — this is the first strategy in the add-on that commands the battery
+*against* the meter on purpose, so it is armed deliberately or not at all.
+
+| option | meaning |
+|---|---|
+| `arbitrage_enabled` | `false` by default. Arms the strategy |
+| `cycle_cost_threshold_eur` | What one full round trip costs per kWh — pack wear plus conversion losses. A spread below this is not traded |
+| `arbitrage_soc_floor` | The reserve arbitrage may not spend, % |
+| `arbitrage_soc_ceiling` | The headroom it may not overfill, % |
+
+**What it needs, and what the add-on refuses.** Arbitrage requires a price
+source (`price_provider` or `static_profile`) and a P1 meter (`meter_source`).
+Without a price source there is nothing to tell cheap from expensive; without
+P1 the capacity-tariff peak guard never runs, and §9.1 puts peak shaving above
+arbitrage — a charge that nothing measures is exactly what that guard exists to
+prevent. Both combinations are **refused at startup** rather than left looking
+armed.
+
+**Peak shaving always wins.** The strategy cannot obtain a spread at all
+without first answering whether the peak guard is holding, and even if it
+could, the guard's floor is filed at a higher arbiter priority and clamps an
+arbitrage charge. Discharging is never capped by it — only offtake is billed.
+
+**It never exports into a negative price** (§17). A negative hour is the best
+possible hour to *charge* in and that is still allowed; selling into one gives
+energy away and pays for the privilege.
+
+**At the shipped threshold a static day/night site does not trade.** The
+realistic Belgian day/night spread is about 0.08 EUR/kWh and the default
+threshold is 0.10, so the honest answer on such a site is "no opportunity", and
+the status page says so with the numbers. Whether 0.10 is the right default is
+a product decision, not a bug: lower it only if you know what a cycle costs
+your pack.
+
+**The prices are fetched in the background**, never on the control tick — a
+supplier's TLS timeout on that path would be a supplier able to push the add-on
+towards the firmware's 30 s failsafe. The plan refreshes every 30 minutes and
+retires itself if the refresh stops answering, at which point arbitrage stops
+with it.
+
 ### Site
 
 | option | meaning |

goodwe_controller/app/arbiter.py

@@ -36,6 +36,15 @@ P_SAFETY = 100        # stopped, inputs missing, shutting down, device limits
 P_MAINTENANCE = 60    # maintenance owns the actuator outright (drain, hold)
 P_PEAK = 50           # capacity-tariff peak shaving - costs real money
 P_MAINT_SHAPE = 40    # maintenance shaping the loop (charge-only)
+# OPT-04. §10.1 priority 3 - price arbitrage. ⚠️ ITS POSITION IS THE SPEC, NOT A
+# preference: P_PEAK (50) is ABOVE it so clause 2 lets the capacity-tariff floor
+# clamp an arbitrage charge, which is §9.1's "give peak shaving priority over
+# arbitrage whenever the two conflict" expressed as arithmetic rather than as a
+# rule somebody has to remember. P_MAINT_SHAPE (40) is above it so a speculative
+# spread cannot override the pack's balancing cycle. P_LOOP (10) is below it
+# because trading against the meter is the entire point - the grid-following
+# loop would otherwise cancel every charge the moment it created import.
+P_ARBITRAGE = 30      # price arbitrage - trades against the meter on purpose
 P_LOOP = 10           # ordinary grid-following
 
 SET = "set"

goodwe_controller/app/arbitrage.py

@@ -0,0 +1,429 @@
+"""OPT-04 - §10.1 priority 3: charge cheap, discharge expensive, and only when
+the round trip is worth more than it costs.
+
+WHERE THIS LIVES, AND WHY IT IS NOT WHERE THE CARD SAYS. The functional notes
+on the Trello card reference specs.md §2.2's `ems/optimizer/...` layout. That
+module layout does not exist and is not being built; the product owner's
+decision of 2026-08-30 is that everything ships in `goodwe_controller/app/`,
+beside price.py, static_tariff.py, peakguard.py and arbiter.py. The BEHAVIOUR
+§10.1 / §9.1 / §8.3 / §17 describes is unchanged and still authoritative.
+
+
+WHAT THIS MODULE IS, AND - MORE IMPORTANTLY - WHAT IT IS NOT
+============================================================
+Four cards landed before this one and this one CONSUMES all four. Almost every
+question an arbitrage strategy has to answer is already answered somewhere
+else, by an object built for it, and the single biggest way to get this ticket
+wrong is to answer any of them again here:
+
+  price.py (TARIFF-02)          what a price IS, all-in, and what a product is.
+                                `Uncertainty`, `WIDEN_FACTOR`, `required_spread`.
+  static_tariff.py (TARIFF-03)  `TariffPlan.arbitrage()` - THE VERDICT. Whether
+                                a trade is permitted at all, on which two hours,
+                                against which widened threshold. Five refusal
+                                reasons live there, not here.
+  price_forecast.py (FCST-03)   provisional carrying, and the rule that no
+                                irreversible action rests on a provisional
+                                price (§8.3).
+  peakguard.py (SAFETY-07)      the capacity-tariff floor, filed as a `lo`
+                                LIMIT at P_PEAK on the arbiter.
+
+So what is left for this file is genuinely small, and that is the point:
+
+  * WHICH HOUR IS IT NOW - is this instant the buy hour, the sell hour, or
+    neither.
+  * MAY THE BATTERY DO IT - state of charge, and §17's negative-price rule.
+  * HOW HARD - one number, filed as one arbiter claim.
+
+Everything else is delegated, and delegated by CONSTRUCTION rather than by
+convention: `TariffPlan.arbitrage()` will not hand over a spread at all unless
+it is told whether the peak guard is holding, what a cycle costs, and what is
+uncertain. There is no code path here that can reach a tradeable number without
+having answered §10.1's priority ordering first.
+
+
+AC3 - PEAK SHAVING WINS, AND IT WINS TWICE
+==========================================
+§9.1: "Give peak shaving priority over arbitrage whenever the two conflict."
+That is honoured through the two mechanisms that already exist, and NEITHER of
+them is re-implemented here:
+
+  1. THE INTERLOCK. `plan.arbitrage(peak_guard_active=...)` is keyword-only and
+     REQUIRED. When the guard is engaged the verdict is a refusal carrying the
+     §10.1 wording, so `decide()` below never sees a spread at all. It is not
+     possible to write a caller here that skips the question - the argument has
+     no default and the spread is unreachable without it.
+
+  2. THE ARBITER. Even if (1) were somehow bypassed, `claim()` files a SET at
+     P_ARBITRAGE (30) and peakguard files a LIMIT at P_PEAK (50). arbiter.py
+     clause 2 applies every limit whose priority is >= the winning set's, most
+     restrictive wins - so the peak floor CLAMPS an arbitrage charge whatever
+     this module thinks. 50 >= 30 is what makes that true, and it is why the
+     priority below sits under P_PEAK rather than beside it.
+
+Two independent mechanisms for one rule is deliberate. The first is the one
+that reports honestly; the second is the one that holds when the first is
+wrong.
+
+⚠️ AND THE GUARD MUST BE ABLE TO SEE. On `meter_source: off` there is no
+telegram stream, so peakguard.decide() is never run and `peak_guard_active` is
+not FALSE - it is UNKNOWN. Charging a battery at max_w creates offtake, offtake
+is the billed quantity (§9.1), and doing it deliberately while nothing measures
+the quarter is the exact shape of failure §9.1 exists to prevent. So the third
+state is carried explicitly as `None` and it REFUSES. A failed read is not an
+observation, and "the guard did not run" is not "the guard is clear".
+
+
+§17's NEGATIVE-PRICE CASE, STATED ONCE
+======================================
+"Negative-price hours - arbitrage must not export." A negative price means the
+market pays the site to consume. Discharging the battery into such an hour
+gives away stored energy AND forgoes being paid to take more in; if the house
+load is covered it becomes export, which at a negative price is paid-for
+disposal. The buy side is untouched - a negative hour is the best possible hour
+to charge in, and this file's guard is deliberately asymmetric for that reason.
+
+⚠️ THE GUARD IS ON THE HOUR BEING SOLD INTO, NOT ON THE SPREAD. A series whose
+prices are -0.30 and -0.05 has a 0.25 EUR/kWh spread that clears any shipped
+threshold, and its DEAREST hour is still negative. A guard written as
+"spread > 0" or "the prices are not all negative" passes that case and exports.
+
+
+WHAT `cycle_cost_threshold_eur` IS, AND WHAT THIS FILE REFUSES TO DO WITH IT
+===========================================================================
+§10.1 gates arbitrage on "expected round-trip benefit exceeds
+`cycle_cost_threshold_eur`". It ships at 0.10 EUR/kWh. On realistic Belgian
+day/night numbers the static spread is about 0.08 EUR/kWh, and FCST-03's
+widening factors (2.5 real-time, 1.5 no-weather) are honest policy defaults
+with nothing to derive them from.
+
+⚠️ SO A STATIC-TARIFF SITE DOES NOT TRADE AT SHIPPED DEFAULTS, AND THAT IS
+REPORTED RATHER THAN TUNED AWAY. The threshold is a parameter here, the
+behaviour is correct at any value of it, and no number in this file was chosen
+to make a trade happen. Whether 0.10 is the right shipped default - and whether
+2.5/1.5 are - is one product decision and it belongs to the owner, not to this
+card. test_arbitrage.py section 7 reports what the shipped default does over a
+replayed year rather than adjusting anything so that it looks better.
+
+
+THE ACCOUNTING BASIS, AND THE TERM THAT IS DELIBERATELY ABSENT
+=============================================================
+The card's technical notes ask for `benefit = value_out(d) - cost_in(c)/η`, and
+then say of the round-trip efficiency: "η is NOT a §14 key - it needs one, or
+must be derived from measured charge/discharge energy in the state store. DO
+NOT HARDCODE IT." Both halves are honoured by not having the term at all:
+
+  * THE SELL SIDE IS PRICED AS DISPLACED IMPORT, NOT AS EXPORT. The two hours
+    compared are both all-in CONSUMPTION prices, which is the basis the notes
+    require ("credit displacement at all-in import prices, not at export price,
+    or the number flatters itself"). `export_price_eur_kwh` is therefore not
+    read here at all - §9.3's near-zero Belgian injection price is precisely
+    why the economically sound trade on this market is charge-cheap-to-
+    displace-expensive-import, and §17's guard below is the same fact stated as
+    a rule.
+
+  * AND THERE IS NO η, BECAUSE THERE IS NOTHING HONEST TO PUT IN IT. No config
+    key exists, the state store carries no measured charge/discharge energy
+    pair to derive one from, and a hardcoded 0.85 is exactly what the notes
+    forbid - it would also DOUBLE-COUNT, because `cycle_cost_threshold_eur` is
+    already defined as what a round trip costs per kWh and conversion loss is
+    part of that cost. When a measured η exists it belongs beside the threshold
+    as a second term, and this file's arithmetic is one subtraction, in one
+    place, ready for it.
+
+
+HYSTERESIS (§10.2)
+==================
+The switching decision is "which hour is it", and an hour is 60 minutes wide,
+so the action cannot chatter within one. The other two terms are monotone
+inside an hour in the only direction that matters: a charge raises SoC towards
+the ceiling and a discharge lowers it towards the floor, so each can stop once
+and cannot restart in the same hour. No timer is needed and none is invented.
+"""
+
+import logging
+import math
+from dataclasses import dataclass
+from datetime import datetime
+
+from .arbiter import P_ARBITRAGE, Claim
+from .price import Absence, price_option
+
+_LOG = logging.getLogger("goodwe.arbitrage")
+
+SOURCE = "arbitrage"
+
+# §10.2 reason codes. Written out because the action log is a user-facing
+# artefact and a code assembled from an f-string cannot be grepped for.
+CHARGE = "ARBITRAGE_CHARGE"
+DISCHARGE = "ARBITRAGE_DISCHARGE"
+
+# Why nothing was done. These are the reasons this MODULE owns; the five
+# refusals that belong to the verdict itself are carried through verbatim from
+# static_tariff.py rather than restated, so a refusal's wording has exactly one
+# author.
+IDLE_NO_PLAN = "no tariff plan has been resolved yet"
+IDLE_NOT_COVERED = (
+    "NOT COVERED: meter_source is off, so the capacity-tariff peak guard never "
+    "runs and nothing measures the quarter this charge would land in. §9.1 puts "
+    "peak shaving above arbitrage, and an unmeasured peak cannot be shaved - "
+    "so the honest answer is that arbitrage is not covered on this install, not "
+    "that the guard is clear")
+IDLE_NO_SOC = (
+    "state of charge is not being read, and a battery whose fill nobody knows "
+    "cannot be told to empty or to fill")
+IDLE_OFF_PEAK_HOUR = "this hour is neither the buy hour nor the sell hour"
+NEGATIVE_PRICE = (
+    "§17: the sell hour is priced BELOW ZERO, so discharging into it would give "
+    "energy away and pay for the privilege. Arbitrage must not export at a "
+    "negative price")
+
+
+@dataclass(frozen=True)
+class ArbitrageDecision:
+    """What the strategy concluded this cycle, with what it concluded it from.
+
+    ⚠️ NO FIELD HAS A DEFAULT, and that is TEST-14's rule rather than a style
+    choice: an outcome type's required field quietly gaining a default is a
+    mutation that survived review in three files in this repository. The
+    signature is scanned in test_price_forecast.py alongside `Gate`'s and
+    `Arbitrage`'s constructors, so a default added here later goes red without
+    anybody remembering to write a case for it.
+
+    `action` is None whenever nothing is to be done, and `power_w` is then None
+    too - not 0.0. Zero watts is a COMMAND (it parks the battery); "no claim"
+    is the absence of one, and the arbiter must be free to let the loop win.
+    """
+
+    action: str | None            # CHARGE | DISCHARGE | None
+    power_w: float | None         # signed: negative charges, positive discharges
+    reason: str
+    spread_eur_kwh: float | None
+    required_eur_kwh: float | None
+    price_now_eur_kwh: float | None
+
+    @property
+    def trading(self) -> bool:
+        return self.action is not None
+
+    def explain(self) -> str:
+        if not self.trading:
+            return f"arbitrage idle - {self.reason}"
+        return self.reason
+
+
+def _idle(reason, *, spread=None, required=None, price=None) -> ArbitrageDecision:
+    if not reason or not isinstance(reason, str):
+        raise ValueError(
+            "an idle arbitrage decision needs a reason a human can act on - "
+            "§10.2 requires every plan change to carry one, and 'arbitrage did "
+            "nothing' with no reason is indistinguishable from a bug.")
+    return ArbitrageDecision(action=None, power_w=None, reason=reason,
+                             spread_eur_kwh=spread, required_eur_kwh=required,
+                             price_now_eur_kwh=price)
+
+
+def enabled(opts: dict) -> bool:
+    """Is the strategy switched on at all? Off in the shipped configuration.
+
+    ⚠️ OFF BY DEFAULT ON PURPOSE. Every other card in this repository ships its
+    machinery inert and lets an operator arm it; this one MOVES ENERGY against
+    the meter on a real battery, so it is the last one that should be an
+    exception to that.
+    """
+    v = price_option(opts, "arbitrage_enabled")
+    return v is True or str(v).strip().lower() in ("true", "yes", "on", "1")
+
+
+def threshold_eur_kwh(opts: dict) -> float:
+    """§10.1's `cycle_cost_threshold_eur`, from the one place option defaults live."""
+    v = float(price_option(opts, "cycle_cost_threshold_eur"))
+    if not math.isfinite(v) or v < 0:
+        raise ValueError(
+            f"cycle_cost_threshold_eur is {v!r}. It is what a round trip COSTS, "
+            f"in EUR/kWh, so it is non-negative and finite - a negative "
+            f"threshold is one every spread clears, which is arbitrage with the "
+            f"brake removed.")
+    return v
+
+
+def soc_bounds(opts: dict) -> tuple:
+    """(floor, ceiling) in percent. The reserve arbitrage may not spend."""
+    return (float(price_option(opts, "arbitrage_soc_floor")),
+            float(price_option(opts, "arbitrage_soc_ceiling")))
+
+
+def guard_state(peak_decision):
+    """The peak guard's answer as the three-state `decide()` takes: T / F / None.
+
+    THE ONE PLACE THIS CONJUNCTION IS WRITTEN, and it is public for the reason
+    `peakguard.shaping_suspended()`, `watchdog.loop_is_following()` and
+    `freeze.looks_frozen()` are: the check that grades it has to assert on THE
+    SHIPPED EXPRESSION. A predicate transcribed into its own test is one copy
+    plus a future disagreement about what the check means, and the copy that
+    drifts is always the grading one - this repository has three separate
+    records of exactly that, one of which left every suite green while a
+    default install parked at 0 W in the quarter that sets the month's bill.
+
+    ⚠️ `None` IS NOT `False` AND COLLAPSING THE TWO IS THE WHOLE HAZARD. On
+    `meter_source: off` main.py never runs peakguard.decide() - there is no
+    telegram stream, so there is no open quarter to project - and the decision
+    it hands here is None. That is "the guard never ran", not "the guard is
+    clear", and answering False would arm a strategy whose charging creates the
+    billed quantity (§9.1) while nothing measures the quarter it lands in.
+    """
+    if peak_decision is None:
+        return None
+    return bool(peak_decision.engaged)
+
+
+def _point_at(plan, when: datetime):
+    """The price point covering `when`, or None when the series does not cover it.
+
+    ⚠️ None IS NOT A PRICE AND IS NOT ZERO. `TariffPlan` raises `Absence` here
+    rather than answering 0.0, for price.py's reason - "no data" becoming "free"
+    is the defect the whole outcome-type family exists to remove - and this
+    turns that into "do nothing", never into "it is cheap".
+    """
+    try:
+        return plan.read.match(series=lambda r: r.at(when),
+                               no_data=lambda _r: None)
+    except Absence:
+        return None
+
+
+def decide(plan, now: datetime, *, peak_guard_active, soc,
+           min_spread_eur_kwh, uncertainty, charge_w, discharge_w,
+           soc_floor_pct, soc_ceiling_pct) -> ArbitrageDecision:
+    """Should the battery trade this cycle, and how hard.
+
+    ⚠️ `peak_guard_active` IS A THREE-STATE ARGUMENT AND ALL THREE ARE REAL.
+    True (the guard is holding), False (it ran and is clear), and None (it never
+    ran, because nothing measures this install's quarter). See the module note:
+    None refuses, because the alternative is a strategy that creates the billed
+    quantity while the thing that would have limited it is switched off.
+
+    ⚠️ EVERY OTHER ARGUMENT IS KEYWORD-ONLY AND HAS NO DEFAULT, for
+    `TariffPlan.arbitrage()`'s reason. A defaulted `soc_floor_pct` is a reserve
+    floor this module asserted rather than one the operator configured, and the
+    first time it is wrong the battery is at 0 %.
+    """
+    if peak_guard_active is None:
+        return _idle(IDLE_NOT_COVERED)
+    if plan is None:
+        return _idle(IDLE_NO_PLAN)
+
+    verdict = plan.arbitrage(peak_guard_active=peak_guard_active,
+                             min_spread_eur_kwh=min_spread_eur_kwh,
+                             uncertainty=uncertainty)
+    return verdict.match(
+        refused=lambda v: _idle(v.reason),
+        allowed=lambda v: _act(plan, now, v, soc, charge_w, discharge_w,
+                               soc_floor_pct, soc_ceiling_pct))
+
+
+def _act(plan, now, v, soc, charge_w, discharge_w, floor_pct, ceiling_pct):
+    """The permitted case: which hour is this, and may the pack do it."""
+    here = _point_at(plan, now)
+    if here is None:
+        return _idle(
+            f"the series carries a tradeable spread but no price for this "
+            f"instant ({now.isoformat()}), which is §11.4 for this hour rather "
+            f"than a reason to reach for a neighbouring one",
+            spread=v.spread_eur_kwh, required=v.required_spread_eur_kwh)
+
+    price = here.eur_per_kwh
+    buying = here.start == v.cheapest.start
+    selling = here.start == v.dearest.start
+    if not (buying or selling):
+        return _idle(IDLE_OFF_PEAK_HOUR, spread=v.spread_eur_kwh,
+                     required=v.required_spread_eur_kwh, price=price)
+
+    # ⚠️ BELT AND BRACES ON §8.3, AND IT IS NOT DEAD CODE. `_firm_or_die` makes
+    # the two points on the verdict firm at construction, but the point matched
+    # HERE is whatever `at()` returned for this instant, and the identity of the
+    # two rests on the series being the one the verdict was built from. An hour
+    # that is provisional is one nobody published a price for, and no
+    # irreversible action may rest on it.
+    if here.provisional:
+        return _idle(
+            f"§8.3: the price for this hour ({here.start.isoformat()}) is "
+            f"CARRIED FORWARD, not published, and no irreversible action may be "
+            f"taken on a provisional price",
+            spread=v.spread_eur_kwh, required=v.required_spread_eur_kwh,
+            price=price)
+
+    if soc is None:
+        return _idle(IDLE_NO_SOC, spread=v.spread_eur_kwh,
+                     required=v.required_spread_eur_kwh, price=price)
+
+    common = dict(spread=v.spread_eur_kwh, required=v.required_spread_eur_kwh,
+                  price=price)
+
+    if selling:
+        # §17. THE NAMED REGRESSION CASE. Checked BEFORE the SoC question, so a
+        # negative price reports the reason that actually forbids the trade
+        # rather than one that happens to fire first on some other site.
+        if price < 0:
+            return _idle(f"{NEGATIVE_PRICE} - this hour is "
+                         f"{price:+.4f} EUR/kWh", **common)
+        if soc <= floor_pct:
+            return _idle(
+                f"the sell hour is here at {price:.4f} EUR/kWh but the pack is "
+                f"at {soc:g} %, at or under the {floor_pct:g} % reserve floor "
+                f"arbitrage may not spend (§10.1 priority 4)", **common)
+        return ArbitrageDecision(
+            action=DISCHARGE, power_w=abs(float(discharge_w)),
+            reason=(f"{DISCHARGE}: this is the dearest firm hour at "
+                    f"{price:.4f} EUR/kWh, {v.spread_eur_kwh:.4f} above the "
+                    f"cheapest and clear of the {v.required_spread_eur_kwh:.4f} "
+                    f"this cycle had to cover; discharging "
+                    f"{abs(float(discharge_w)):.0f} W from {soc:g} %"),
+            spread_eur_kwh=v.spread_eur_kwh,
+            required_eur_kwh=v.required_spread_eur_kwh,
+            price_now_eur_kwh=price)
+
+    if soc >= ceiling_pct:
+        return _idle(
+            f"the buy hour is here at {price:.4f} EUR/kWh but the pack is at "
+            f"{soc:g} %, at or over the {ceiling_pct:g} % ceiling - there is "
+            f"nowhere to put the cheap energy", **common)
+    return ArbitrageDecision(
+        action=CHARGE, power_w=-abs(float(charge_w)),
+        reason=(f"{CHARGE}: this is the cheapest firm hour at "
+                f"{price:.4f} EUR/kWh, {v.spread_eur_kwh:.4f} below the dearest "
+                f"and clear of the {v.required_spread_eur_kwh:.4f} this cycle "
+                f"had to cover; charging {abs(float(charge_w)):.0f} W from "
+                f"{soc:g} %"),
+        spread_eur_kwh=v.spread_eur_kwh,
+        required_eur_kwh=v.required_spread_eur_kwh,
+        price_now_eur_kwh=price)
+
+
+def claim(decision: ArbitrageDecision) -> Claim | None:
+    """The decision as one arbiter claim, or None when nothing is to be traded.
+
+    A SET at P_ARBITRAGE, and the priority is the whole §10.1 ordering:
+
+      P_PEAK 50        > this, so peakguard's `lo` floor CLAMPS an arbitrage
+                         charge (arbiter clause 2 applies every limit at or
+                         above the winner's priority). AC3, mechanically.
+      P_MAINT_SHAPE 40 > this, so a maintenance charge window's `hi` shaping is
+                         not overridden by a price opportunity. §12.4's "money
+                         outranks the schedule" is about the PEAK guard, which
+                         is measured and billed; a speculative spread is not
+                         that, and letting it out-rank the pack's balancing
+                         cycle would be this card quietly rewriting §12.
+      P_LOOP 10        < this, because trading against the meter is the point:
+                         the grid-following loop would otherwise cancel every
+                         charge the moment it created import.
+
+    ⚠️ NONE MEANS NO CLAIM, NOT A 0 W CLAIM. Filing `Claim.set(..., 0.0)` here
+    would out-rank the loop with a park order every cycle the strategy had
+    nothing to say, i.e. it would silently disable self-consumption on any site
+    that armed arbitrage.
+    """
+    if not decision.trading or decision.power_w is None:
+        return None
+    return Claim.set(SOURCE, P_ARBITRAGE, float(decision.power_w),
+                     decision.action)

goodwe_controller/app/main.py

@@ -28,10 +28,11 @@ import contextlib
 import json
 import logging
 import signal
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 import aiohttp
 
+from . import arbitrage
 from .arbiter import (
     P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_PEAK, P_SAFETY, Claim, resolve,
 )
@@ -48,7 +49,8 @@ from .maintenance import IDLE, MaintConfig, Maintenance
 from .mqtt import MqttPublisher
 from . import peakguard
 from .p1 import P1Ingest, build_source, is_enabled
-from .static_tariff import static_status
+from .price import Uncertainty, price_option
+from .static_tariff import resolve_tariff, static_status
 from .tariff import PeakStore, PeakTracker
 from .validate import ConfigRefused, refuse, validate_startup_config
 from .watchdog import SignWatchdog, SignWatchdogConfig, loop_is_following
@@ -57,6 +59,34 @@ from . import web
 OPTIONS_PATH = "/data/options.json"
 _LOG = logging.getLogger("goodwe")
 
+# --------------------------------------------------------------------------- #
+# OPT-04: THE BACKGROUND TARIFF REFRESH, AND WHY IT IS NOT OPTIONAL
+#
+# `static_tariff.resolve_tariff()` may perform an HTTPS fetch. The control tick
+# is what feeds the ESP32 heartbeat, and layer 1 commands 0 W if that heartbeat
+# stops for ~30 s - so a supplier's TLS timeout on the tick is a SUPPLIER able
+# to push this add-on into the failsafe. Every module in the price stack says
+# so in its own words (static_tariff.static_plan, resolve_tariff, main's
+# read_inputs) and this is the task those notes were waiting for.
+#
+# The tick therefore never awaits a fetch. It reads `self.tariff_plan`, an
+# attribute this task assigns, and nothing else.
+#
+# ⚠️ THE CADENCE IS NOT A GUESS AT HOW OFTEN PRICES CHANGE. A day-ahead feed
+# publishes ONCE, at 14-16 h CET, and §8.3's carried hours become firm at that
+# moment; 30 minutes is short enough that the switch is picked up inside the
+# hour it happens and long enough to be a polite 48 fetches a day. Nothing in
+# the control law depends on it.
+TARIFF_REFRESH_S = 1800.0
+# ⚠️ AND THE AGE BOUND IS NOT THE SAME QUESTION. A refresh that FAILS still
+# stores a plan - resolve_tariff() degrades to the §11.4 static fallback, which
+# is provisional and refuses arbitrage on its own. This bound covers the other
+# case: the refresh task itself dying, leaving a plan that was true hours ago
+# standing as if it were current. Three missed refreshes is the threshold, for
+# `p1.max_age_s`'s reason - one missed poll is not a fault, a run of them is.
+# A failed read is not an observation, and neither is a stale one.
+TARIFF_MAX_AGE_S = 3 * TARIFF_REFRESH_S
+
 
 def load_options() -> dict:
     try:
@@ -199,14 +229,25 @@ class Controller:
         # the action log and the notification fire on transitions rather than
         # once per cycle for a quarter of an hour.
         #
-        # ⚠️ NO CONFIG OF ITS OWN, for the reason the clamp watchdog gives ~90
-        # lines up and TARIFF-01 repeats: the ceiling is `peak_cap_w`, already
-        # shipped and already validated, and every other term is read off
-        # something the arbiter or the averager already owns. CFG-03 still
-        # tracks three places, not four.
+        # ⚠️ THE CEILING HAS NO CONFIG OF ITS OWN, for the reason the clamp
+        # watchdog gives ~90 lines up and TARIFF-01 repeats: it is `peak_cap_w`,
+        # already shipped and already validated, and every other term of the
+        # projection is read off something the arbiter or the averager already
+        # owns.
+        #
+        # ⚠️ SAFETY-24'S FLOOR DOES HAVE ONE, AND IT COULD NOT NOT HAVE. How far
+        # down the guard may spend the pack for money is a number only the site
+        # knows, and refusing one set below the pack's own cut-out means knowing
+        # that cut-out too. So this card adds exactly two keys and reads their
+        # defaults from `peakguard.PEAK_OPTION_DEFAULTS` - TARIFF-02's mirror
+        # pattern - rather than typing literals here. THAT is what keeps CFG-03
+        # at three places: the divergence validate.py's rule 6 documents
+        # (`maintenance_charge_w` 2000 there, 2500 here) is exactly what a
+        # literal in this file would repeat.
         self.peak_guard = None
         self._peak_engaged = False
         self._peak_saturated = False
+        self._peak_floor_limited = False
         self._maint_shape_suspended = False
         # Set by amain() once the transport is built, so the status page can show
         # what only the transport knows (homewizard_local's unchanged_s). Stays
@@ -232,6 +273,16 @@ class Controller:
         # window we are in and why it is or is not cheap - so "Cheap window:
         # no" is an evidenced answer rather than a bare no.
         self.tariff_words = "no static profile"
+
+        # OPT-04. The tariff plan the BACKGROUND task last resolved, and when.
+        # cycle() reads these two attributes and never awaits a fetch - see
+        # TARIFF_REFRESH_S. `None` means no refresh has completed yet, which is
+        # not "no prices" and certainly not "free": arbitrage.decide() reports
+        # it as its own reason and does nothing.
+        self.tariff_plan = None
+        self.tariff_at = None
+        self.arb = None
+        self._arb_action = None
         self.last_write = None
         self.last_write_at = None
         self.last_write_ok = None
@@ -494,6 +545,17 @@ class Controller:
             cap_w=float(self.o.get("peak_cap_w", 3500)),
             reading=self.peak.peak(now),
             achievable_w=min(max_w, abs(float(dev_top))),
+            # ⚠️ SAFETY-24. `self.soc` IS THE MEASURED READING AND MAY BE None,
+            # and it is handed over as None rather than defaulted here. An early
+            # draft of the stage-5 loop read SoC as `| float(100)` and thereby
+            # invented a full battery (NOTES.md 4.4b); a full battery is
+            # precisely the value that disarms this floor. decide() treats None
+            # as floor-limited, which is the direction that costs a shave rather
+            # than the pack.
+            soc=self.soc,
+            soc_floor=float(self.o.get(
+                "peak_soc_floor",
+                peakguard.PEAK_OPTION_DEFAULTS["peak_soc_floor"])),
             max_hold_s=self.p1.averager.max_hold_s,
             frozen=looks_frozen(getattr(self.p1_source, "unchanged_s", None),
                                 standing, max_w, self.freeze_wd.cfg),
@@ -514,6 +576,64 @@ class Controller:
             if self.peak_guard is not None:
                 self.log_event(self.peak_guard.explain())
 
+        # --- OPT-04: price arbitrage (§10.1 priority 3) ----------------------
+        # ONE `set` claim at P_ARBITRAGE, and app/arbitrage.py carries the whole
+        # argument. What is decided HERE and nowhere else is what the strategy
+        # is allowed to see:
+        #
+        # ⚠️ IT IS PLACED AFTER THE PEAK BLOCK BECAUSE IT READS ITS CONCLUSION.
+        # `engaged_now` is this cycle's peak verdict, computed ~20 lines up.
+        # Evaluated before it, this would hand the strategy the PREVIOUS cycle's
+        # answer - one cycle of lag on the interlock that implements §9.1's
+        # "peak shaving outranks arbitrage", inside a 900 s quarter.
+        #
+        # ⚠️ AND THE THIRD STATE IS NOT A TIDINESS. Without P1 the guard is not
+        # CLEAR, it never RAN, and `None` says so. arbitrage.decide() refuses on
+        # it, because charging at max_w creates the billed quantity while
+        # nothing is measuring the quarter it lands in. `False` here would be
+        # this line asserting a measurement that was never taken - the defect
+        # `PeakReading` and `quarter_ceiling_w()` both exist to remove.
+        #
+        # ⚠️ THE THREE-STATE COLLAPSE LIVES IN arbitrage.guard_state(), NOT
+        # HERE, and that is not tidiness. It is the same lesson
+        # `shaping_suspended()` records: inline, this expression was reachable
+        # by no check, and mutating it back to a bare `engaged_now` left every
+        # suite in this repository green while a `meter_source: off` install
+        # traded blind. Read guard_state() before touching this line.
+        #
+        # ⚠️ NO FETCH ON THIS PATH. `fresh_tariff_plan()` is an attribute read
+        # and an age comparison; the HTTPS call lives in run_tariff_refresh().
+        # See TARIFF_REFRESH_S: a supplier's TLS timeout here would be a
+        # supplier able to push this add-on towards layer 1's 30 s veto.
+        #
+        # ⚠️ AND THE UNCERTAINTY SET IS NOT EMPTY BY DEFAULT. §11.4's
+        # no-weather row widens the threshold, and this add-on carries no
+        # weather forecast at all - §8.1's PV forecast is not built. Passing
+        # `frozenset()` would be main.py asserting a forecast is healthy when
+        # there is not one, which is exactly what static_tariff's `uncertainty`
+        # argument has no default in order to prevent.
+        self.arb = None
+        if arbitrage.enabled(self.o):
+            floor_pct, ceiling_pct = arbitrage.soc_bounds(self.o)
+            self.arb = arbitrage.decide(
+                self.fresh_tariff_plan(now), now,
+                peak_guard_active=arbitrage.guard_state(self.peak_guard),
+                soc=self.soc,
+                min_spread_eur_kwh=arbitrage.threshold_eur_kwh(self.o),
+                uncertainty=frozenset({Uncertainty.NO_WEATHER}),
+                charge_w=max_w, discharge_w=max_w,
+                soc_floor_pct=floor_pct, soc_ceiling_pct=ceiling_pct)
+            arb_claim = arbitrage.claim(self.arb)
+            if arb_claim is not None:
+                claims.append(arb_claim)
+            # Edge-logged, for the peak guard's reason: the action log holds 40
+            # lines and a trade lasts a whole hour, so a line per cycle would
+            # flush the context that explains it. The transition is what an
+            # operator needs - "it started buying at 03:00", not 240 copies.
+            if self.arb.action != self._arb_action:
+                self._arb_action = self.arb.action
+                self.log_event(self.arb.explain())
+
         # --- maintenance -----------------------------------------------------
         if result.owns_setpoint:
             claims.append(Claim.set("maintenance", P_MAINTENANCE,
@@ -682,6 +802,16 @@ class Controller:
         # method assigned above and nothing since has touched.
         await self.check_peak_saturated()
 
+        # --- SAFETY-24's alarm, and it is the OTHER way the money gets spent --
+        # Same boundary argument as check_peak_saturated(): it grades
+        # `self.peak_guard`, which cycle() assigned above and nothing since has
+        # touched. Kept as its own method rather than folded into that one,
+        # because the two states are mutually exclusive and want OPPOSITE
+        # operator actions - saturation says "shed load, the inverter is already
+        # flat out"; this says "the pack is at its floor and this quarter is not
+        # being shaved at all".
+        await self.check_peak_floor_limited()
+
         # --- TARIFF-01, and it is LAST on purpose -----------------------------
         # Banking a quarter is bookkeeping about a moment that has already
         # happened; a slow SD card must never sit between the meter and the
@@ -826,6 +956,70 @@ class Controller:
                 f"thing that changes the bill."),
         })
 
+    async def check_peak_floor_limited(self) -> None:
+        """SAFETY-24. Alarm when the EMS SoC floor stops the guard shaving.
+
+        ⚠️ FAILING TO SHAVE MUST NOT BE SILENT, AND THAT IS THE WHOLE POINT.
+        The guard exists for one euro figure - the capaciteitstarief bills the
+        month's single worst quarter - so a cycle in which it CHOSE not to shave
+        is a cycle in which money is being spent on purpose. Under-shaving
+        quietly would leave the operator with a bill and a status page full of
+        green.
+
+        The distinction from check_peak_saturated() is the operator's action,
+        which is why these are two notifications and not one:
+
+          saturated      the guard asked for everything the inverter has and it
+                         was not enough. SHED LOAD - the controller has already
+                         done all it can, and it is still discharging.
+          floor-limited  the guard asked for NOTHING, because the pack is at or
+                         under the EMS floor, or its SoC could not be read.
+                         Nothing is being discharged for this quarter at all.
+
+        They cannot both be true: peakguard.decide()'s floor-limited branch
+        returns before the `achievable_w` clamp that sets `saturated`.
+
+        Edge-triggered, and de-asserting arms it again - FreezeWatchdog's shape,
+        for FreezeWatchdog's reason: one notification per episode, not one per
+        cycle for a quarter of an hour.
+        """
+        g = self.peak_guard
+        limited = bool(g is not None and g.floor_limited)
+        was, self._peak_floor_limited = self._peak_floor_limited, limited
+        if not limited or was:
+            return
+        # No prefix: `reason` already opens with PEAK GUARD FLOOR-LIMITED, and
+        # a second copy is the kind of stutter that makes a 40-line action log
+        # read like boilerplate. The saturated path needs its prefix because its
+        # reason carries the word in the middle of a sentence, not at the front.
+        self.log_event(g.reason)
+        _LOG.error(
+            "SAFETY-24 - %s. The controller is STILL RUNNING and charging is "
+            "still blocked, but it is not discharging to hold this quarter, so "
+            "the quarter will set or raise the month's billed capacity peak. "
+            "Shed load, or lower peak_soc_floor if this pack can afford it.",
+            g.reason)
+        # A fixed notification_id, and a DIFFERENT one from every other alarm
+        # here - check_meter_frozen() gives the reasoning. A battery at its
+        # energy floor is not a saturated inverter and is not a frozen meter,
+        # and an operator who has to dismiss three identical-looking cards
+        # learns to dismiss the one that mattered.
+        await self.hass.call("persistent_notification", "create", {
+            "notification_id": "goodwe_peak_soc_floor",
+            "title": "GoodWe: the capacity-peak guard has run out of battery",
+            "message": (
+                f"{g.reason}.\n\nTHIS QUARTER WILL GO OVER THE "
+                f"{g.ceiling_w:.0f} W ceiling and will set or raise the "
+                f"month's billed capaciteitstarief peak, because the peak guard "
+                f"is not allowed to discharge any further: {g.soc_floor_reason}."
+                f" This is not a controller fault - it is the EMS-side energy "
+                f"floor (peak_soc_floor) doing what it is set to do, and it sits "
+                f"deliberately above the pack's own BMS cut-out. Charging stays "
+                f"blocked. Shedding load is the only thing that changes the bill "
+                f"now; if this pack can afford a deeper discharge, lower "
+                f"peak_soc_floor."),
+        })
+
     async def check_meter_frozen(self) -> None:
         """SAFETY-18. Alarm on a meter that answers forever without changing.
 
@@ -866,6 +1060,57 @@ class Controller:
         })
 
     # -- tasks ---------------------------------------------------------------
+    # -- OPT-04: the background tariff refresh -------------------------------
+    def fresh_tariff_plan(self, now):
+        """The stored plan if it is still current, else None. NEVER a stale one.
+
+        ⚠️ THIS IS THE ONE PLACE THE AGE IS JUDGED, and it is public for
+        `loop_is_following()`'s reason: a check has to be able to assert on the
+        shipped expression rather than on a transcription that drifts. Both
+        terms are required - a plan with no timestamp is one nothing stored, and
+        a timestamped plan older than TARIFF_MAX_AGE_S is a reading from a task
+        that has stopped answering.
+        """
+        if self.tariff_plan is None or self.tariff_at is None:
+            return None
+        if (now - self.tariff_at).total_seconds() > TARIFF_MAX_AGE_S:
+            return None
+        return self.tariff_plan
+
+    async def run_tariff_refresh(self) -> None:
+        """Resolve the tariff off the control path, forever. NEVER on the tick.
+
+        ⚠️ IT CATCHES EVERYTHING AND KEEPS GOING, for MqttPublisher's reason:
+        a supplier's outage, a certificate expiry, a malformed feed - all of it
+        is pricing, and the battery does not care. What it must never do is
+        raise out of here and take the task down, because then the plan freezes
+        at its last value and looks current forever. That is what
+        `fresh_tariff_plan()`'s age bound is the backstop for, and this is the
+        half that keeps the task alive so the bound rarely matters.
+
+        ⚠️ AND A FAILURE DOES NOT CLEAR THE PLAN. `resolve_tariff()` already has
+        a §11.4 answer for a dead feed - degrade to the static profile, mark it
+        provisional, arbitrage refuses - and that answer is better than the
+        absence of one. Zeroing the plan here would replace a coarse tariff with
+        no tariff, which is the opposite of what §11.4 asks for.
+        """
+        horizon = float(price_option(self.o, "price_horizon_h"))
+        while not self.stopping:
+            now = datetime.now(timezone.utc)
+            try:
+                plan = await resolve_tariff(
+                    self.o, now, now + timedelta(hours=horizon), now=now)
+                self.tariff_plan, self.tariff_at = plan, now
+                _LOG.info("tariff refreshed: %s", plan.words)
+            except Exception as err:                          # noqa: BLE001
+                _LOG.warning(
+                    "tariff refresh failed (%s); keeping the plan resolved at "
+                    "%s. It retires on its own after %.0f s and arbitrage "
+                    "stops with it.", err, self.tariff_at, TARIFF_MAX_AGE_S)
+            with contextlib.suppress(asyncio.TimeoutError):
+                await asyncio.wait_for(self.stop_event.wait(),
+                                       timeout=TARIFF_REFRESH_S)
+
     async def run_control(self) -> None:
         entity = self.o.get("setpoint_entity", "")
         if entity:
@@ -1116,6 +1361,13 @@ class Controller:
             elif _pg.saturated:
                 out.append({"ok": False, "warn": False,
                             "text": "PEAK GUARD SATURATED - " + _pg.reason})
+            elif _pg.floor_limited:
+                # SAFETY-24. NOT ok, like saturation: in both states the money
+                # is going to be spent and a human is the only actuator left. A
+                # separate row from saturation because the two ask for opposite
+                # things - see check_peak_floor_limited().
+                out.append({"ok": False, "warn": False,
+                            "text": _pg.explain()})
             elif _pg.blind:
                 out.append({"ok": False, "warn": True, "text": _pg.explain()})
             else:
@@ -1243,6 +1495,11 @@ class Controller:
             ("Maintenance due", "now" if next_due is None else next_due.strftime("%Y-%m-%d")),
             ("Cheap window", "yes" if self.cheap else "no"),
             ("Tariff", self.tariff_words),
+            # OPT-04. The strategy's own sentence, not a re-derivation of it -
+            # every refusal it can make already carries a reason a human can
+            # act on, and "Arbitrage: no" would throw all of them away.
+            ("Arbitrage", "off (arbitrage_enabled)" if self.arb is None
+             else self.arb.explain()),
         ]
         return {
             "banner": banner, "level": level, "rows": rows, "checks": checks,
@@ -1370,6 +1627,19 @@ async def amain() -> None:
 
         tasks = [asyncio.create_task(controller.run_control())]
 
+        # OPT-04. The tariff refresh runs as its OWN long-lived task, for the
+        # reason P1 ingestion does and then some: `resolve_tariff()` may open an
+        # HTTPS connection to a supplier, and the control tick is what feeds the
+        # ESP32 heartbeat. A fetch on that path is a supplier's TLS timeout able
+        # to push this add-on towards layer 1's 30 s veto. See TARIFF_REFRESH_S.
+        #
+        # ⚠️ STARTED ONLY WHEN ARBITRAGE IS ARMED. Nothing else in the add-on
+        # reads the resolved plan - `cheap` still comes from static_status() on
+        # the tick, which is pure - so on a default install this task would be
+        # network traffic in aid of an attribute nobody reads.
+        if arbitrage.enabled(opts):
+            tasks.append(asyncio.create_task(controller.run_tariff_refresh()))
+
         # P1 ingestion runs as its own long-lived task. ⚠️ It must not be driven
         # off the control loop: telegrams arrive every ~5 s and the loop would
         # decimate them, so the 15-minute average - the capacity-tariff billing

goodwe_controller/app/peakguard.py

@@ -246,19 +246,100 @@ validate.py are untouched by this ticket, exactly as TARIFF-01 left them, and
 mismatch trap 14's own warning box describes.
 
 ------------------------------------------------------------------------------
-⚠️ WHAT THIS GUARD DOES NOT HAVE: AN ENERGY BACKSTOP. SEE SAFETY-24.
+SAFETY-24: THE ENERGY BACKSTOP THIS GUARD SHIPPED WITHOUT
 
-The graded branch can FORCE DISCHARGE, and nothing in this file or in the loop
-below it consults the state of charge - `soc_floor` belongs to the maintenance
-cycle, and freeze.py already records that for the grid-following loop "the BMS
-is the sole energy backstop". On the 2026-08-20 replay the guard asks for 878 Wh
-in a day. Against a pack sitting near 9 % that is not nothing, and the guard will
-keep asking as long as the meter says the quarter is over the ceiling.
+SAFETY-07 left this file saying, in its own words, that the graded branch can
+FORCE DISCHARGE and that nothing below it consults the state of charge - "the
+BMS is the sole energy backstop" (freeze.py). On the 2026-08-20 replay the guard
+asks for 878 Wh in a day, and it keeps asking for as long as the meter says the
+quarter is over the ceiling. That is the behaviour working as designed, and it
+is unbounded from below by anything in the EMS.
 
-That is deliberate for THIS card and it is filed as SAFETY-24. It is also the
-reason this must not go to a client install before SAFETY-24 lands: on the
-reference site the BMS is a known quantity and an operator is present, and on a
-client site neither is true.
+⚠️ STATE THE RISK FAIRLY, because overstating it is how a real one gets ignored.
+Relying on the BMS is not absurd: it is a genuine floor and it works. But it is
+THE PACK's protection, not the OPERATOR's. A month of aggressive quarters can
+leave the battery repeatedly parked at that floor with no EMS-side record of
+why, and the first symptom - a pack that is never available when it is wanted -
+is indistinguishable from the SoC drift MAINT-01 exists to correct.
+
+So `decide()` now takes `soc` and `soc_floor`, and `discharge_permitted()` is
+the one place the question is answered. Four things about the shape of it:
+
+  1. IT IS NOT A TRIP, AND IT MUST NOT BECOME ONE. SAFETY-07's claim is a LIMIT
+     and every stop in 11.2's table still outranks it. This floor is about
+     ENERGY - how far down the guard may spend the pack for money - not about
+     last-resort safety. The action when it binds is therefore to stop ASKING
+     for more discharge, not to command anything.
+
+  2. THE ACTION IS `floor_w = 0.0`, WHICH IS THE BLIND BRANCH'S ACTION FOR THE
+     BLIND BRANCH'S REASON. Zero is one-directional: it forbids CHARGING, which
+     can only ever lower delivered offtake, and it forbids nothing else. It
+     cannot pull the loop off a discharge it is already doing, because a `lo`
+     limit structurally cannot cap discharge - the property this module's whole
+     shape rests on, and one a NEGATIVE floor would destroy. Nothing here can
+     produce a negative floor: the only two values `floor_w` ever takes are 0.0
+     and `min(wanted, achievable)` with `wanted > 0` already established.
+
+     ⚠️ AND IT DOES NOT BOUND THE LOOP. Self-consumption may still drain the
+     pack past this floor, exactly as SAFETY-18 recorded. That is not fixed
+     here and must not be claimed: this card bounds THE PEAK GUARD's OWN forced
+     discharge, which is the thing SAFETY-07 introduced. A loop-side SoC guard
+     is a different lever with a different right value.
+
+  3. IT IS GRADED ONLY WHERE IT IS NOT DEGENERATE. The SoC question is asked
+     AFTER `wanted > 0` is established - i.e. only where a forced discharge is
+     actually being asked for. Asked earlier it would grade a quantity that
+     governs nothing, and would report "floor-limited" on a quiet afternoon
+     where the guard was never going to discharge anyway.
+
+  4. AN UNREADABLE SoC IS NOT A FULL BATTERY. `soc=None` is floor-limited, not
+     permitted. This is `PeakReading`'s lesson in the one place it bites
+     hardest: a full battery is exactly the value that would disarm this floor,
+     so a failed read defaulting to one would silently restore the unbounded
+     discharge this card exists to remove. Both new arguments are REQUIRED
+     keyword-only and neither has a default, which is TEST-14 read as a rule
+     rather than as a story: a required field that gains a default is how three
+     separate files in this repo shipped a quiet fallback.
+
+AND FAILING TO SHAVE MUST NOT BE SILENT. The whole point of the guard is the
+euro figure, so a floor-limited cycle carries `floor_limited=True` and main.py
+raises ONE notification on the edge, saying the quarter will be exceeded - the
+same shape, and the same argument, as `saturated`. The two are mutually
+exclusive by construction: a floor-limited decision asks for nothing, so it
+cannot have run out of inverter.
+
+------------------------------------------------------------------------------
+CFG-03: TWO NEW OPTIONS, MIRRORED ONCE, AND WHY BOTH ARE NEEDED
+
+SAFETY-07 added none and said so. This card cannot: an EMS-side floor is a
+number only the site knows, and refusing one set below the pack's own cut-out
+means knowing that cut-out too.
+
+  * `peak_soc_floor` - the EMS floor. ⚠️ NOT `maintenance_soc_floor`. That is
+    the maintenance cycle's DRAIN TARGET: the depth a monthly recalibration
+    deliberately goes to, once, under supervision. This is the depth a
+    money-saving discharge may never reach, on any day of the month. Sharing
+    one key would couple two features whose right values differ and would widen
+    CFG-03 rather than narrow it.
+  * `bms_soc_floor` - the pack's own cut-out (register 0x0560 on the reference
+    unit, 10 %). It is here so validate.py rule 8 can REFUSE a `peak_soc_floor`
+    at or below it. That configuration is CFG-01 rule (a): the operator states
+    a floor, the BMS enforces a different one, and the status page reports an
+    energy backstop that can never bind.
+
+Both defaults are mirrored ONCE, in `PEAK_OPTION_DEFAULTS` below, which is
+TARIFF-02's `PRICE_OPTION_DEFAULTS` pattern - imported by validate.py and by
+main.py, pinned to config.yaml by test_peakguard.py rather than transcribed.
+This module still reads no option itself: `decide()` takes both as ARGUMENTS,
+makes no `.get()` call, and the mirror is a default table, not a read.
+
+⚠️ A DECISION RECORDED, NOT A RENAME DONE. Now that a second SoC floor exists,
+`maintenance_soc_floor` is the ambiguous name of the two - it reads like "the
+SoC floor" and is not. Renaming it is an option migration on a live install for
+a clarity gain, so the decision recorded here is: KEEP IT, and carry the
+distinction in DOCS.md and in this comment. Revisit if a THIRD floor appears -
+a loop-side guard would be the third - at which point the migration buys enough
+to be worth its cost.
 
 ------------------------------------------------------------------------------
 AND IT READS NO PER-PHASE QUANTITY, ON PURPOSE
@@ -287,8 +368,44 @@ BLIND_FROZEN = "SAFETY-18: the meter is answering without changing"
 CEILING_CAP = "peak_cap_w"
 CEILING_MONTH_PEAK = "the month's banked peak"
 
+# SAFETY-24. Why the forced discharge may not be sized. Reasons, not booleans,
+# for BLIND_*'s reason: the status page, the action log and the operator
+# notification all have to say WHICH, and "floor-limited" alone is unactionable.
+FLOOR_SOC_UNREADABLE = ("the battery state of charge was not measured this "
+                        "cycle, and an unreadable SoC is not a full battery")
+FLOOR_AT_SOC_FLOOR = "state of charge is at or below the peak guard's SoC floor"
+
 SOURCE = "peak"
 
+# ⚠️ CFG-03. THE ONE IN-CODE MIRROR OF config.yaml FOR THIS CARD'S OPTIONS, and
+# it is TARIFF-02's `PRICE_OPTION_DEFAULTS` pattern for TARIFF-02's reason: the
+# add-on already carries option defaults in three places (config.yaml's
+# `options:`, config.yaml's `schema:`, and main.py's literal `.get()`
+# fallbacks), and a fourth is the open ticket. validate.py and main.py both
+# import THIS dict rather than typing a literal, and test_peakguard.py pins it
+# to config.yaml by READING the file - so a drift goes red instead of shipping.
+#
+# ⚠️ NOT A READ. `decide()` takes both values as arguments and this module makes
+# no `.get()` call at all; the dict is a default table for its two callers.
+PEAK_OPTION_DEFAULTS = {
+    # The EMS-side floor on SAFETY-07's forced discharge, in percent.
+    #
+    # 20 is DERIVED from the two numbers this repository already has, not
+    # chosen. The reference pack is 300 Ah at 51.2 V with DoD 90 - 13.8 kWh
+    # usable, the figure validate.py's rule 6 note already quotes - and the
+    # 2026-08-20 replay models 878 Wh of guard discharge in a day, which is
+    # 6.4 % of that. A floor 10 points above the pack's own 10 % cut-out
+    # therefore leaves the guard roughly 1.6 of its worst observed days of room
+    # before it stops, which is the smallest margin that lets the guard do its
+    # job on a bad day and still never reach the BMS.
+    "peak_soc_floor": 20,
+    # The pack's OWN cut-out, in percent: what the BMS enforces regardless of
+    # anything here. On the reference unit this is register 0x0560, read as 10,
+    # and it is consistent with DoD 90 on a full pack. It exists so validate.py
+    # rule 8 can refuse a `peak_soc_floor` at or below it - see that rule.
+    "bms_soc_floor": 10,
+}
+
 
 @dataclass(frozen=True)
 class PeakDecision:
@@ -309,9 +426,19 @@ class PeakDecision:
     excess_w: float = 0.0
     wanted_w: float | None = None    # the floor before `achievable_w` clamped it
     saturated: bool = False
+    # SAFETY-24. True when a forced discharge WAS wanted and the EMS-side SoC
+    # floor refused to size it. Mutually exclusive with `saturated` by
+    # construction: this branch asks for nothing, so it cannot have run out of
+    # inverter. `soc_floor_reason` says WHICH of the two ways it bound.
+    floor_limited: bool = False
+    soc: float | None = None
+    soc_floor: float | None = None
+    soc_floor_reason: str | None = None
     reason: str = ""
 
     def explain(self) -> str:
+        if self.floor_limited:
+            return self.reason
         if not self.engaged:
             if self.projected_w is None:
                 return f"peak guard idle - {self.reason}"
@@ -383,7 +510,44 @@ def blindness(projection, max_hold_s, frozen: bool) -> str | None:
     return None
 
 
-def decide(projection, *, batt_w, cap_w, reading, achievable_w,
+def discharge_permitted(soc, soc_floor) -> str | None:
+    """SAFETY-24. May the guard size a forced discharge, or why may it not?
+
+    Returns None when it may, and otherwise the REASON it may not - which the
+    decision, the status page, the action log and the operator notification all
+    carry, because "floor-limited" on its own tells nobody what to do.
+
+    THE ONE PLACE THIS QUESTION IS ANSWERED, and it is public for the reason
+    `freeze.looks_frozen()`, `watchdog.loop_is_following()` and
+    `shaping_suspended()` are: a safety predicate transcribed into its own check
+    is one copy plus a future disagreement about what the check means, and the
+    copy that drifts is always the grading one.
+
+    ⚠️ `soc is None` IS NOT PERMISSION. A failed read is not an observation, and
+    a FULL BATTERY is exactly the value that would disarm this floor - so a
+    `float(soc or 100)` here would restore the unbounded discharge this card
+    removes, silently, on the one install whose SoC sensor is misnamed. There is
+    deliberately no default argument on either parameter.
+
+    ⚠️ AND `soc_floor is None` IS NOT PERMISSION EITHER, for the same reason one
+    step further out: a caller that has no floor configured has not thereby been
+    granted an unbounded one.
+
+    Boundary: `<=`, not `<`. A pack sitting exactly ON the floor has already
+    spent everything this guard was allowed to spend; permitting one more cycle
+    there would make the floor mean "one cycle below the floor".
+    """
+    if soc is None:
+        return FLOOR_SOC_UNREADABLE
+    if soc_floor is None:
+        return ("no peak guard SoC floor is configured, so there is nothing to "
+                "bound a forced discharge with")
+    if float(soc) <= float(soc_floor):
+        return FLOOR_AT_SOC_FLOOR
+    return None
+
+
+def decide(projection, *, batt_w, cap_w, reading, achievable_w, soc, soc_floor,
            max_hold_s=None, frozen: bool = False) -> PeakDecision:
     """The whole guard. Pure: no clock, no I/O, no state carried between calls.
 
@@ -395,6 +559,13 @@ def decide(projection, *, batt_w, cap_w, reading, achievable_w,
                    claim can NEVER produce `lo > hi` in the resolver - a
                    contradiction there resolves to 0 W, which would park the
                    battery in the exact minute it should be discharging hardest.
+    `soc`          SAFETY-24. MEASURED state of charge, percent. None when the
+                   sensor for it did not answer - which is floor-limited, not
+                   permission.
+    `soc_floor`    SAFETY-24. The percent below which this guard may not spend
+                   the pack for money. REQUIRED, and deliberately without a
+                   default: a default here would be CFG-03's fourth defaults
+                   site and TEST-14's quiet fallback in one line.
     """
     ceiling, from_where = quarter_ceiling_w(cap_w, reading)
     ceil_top = abs(float(achievable_w))
@@ -454,12 +625,41 @@ def decide(projection, *, batt_w, cap_w, reading, achievable_w,
             wanted_w=wanted,
             reason="the load is under the level this quarter can afford")
 
+    # SAFETY-24 - AND ONLY HERE. `wanted > 0` is now established, so a forced
+    # discharge really is being asked for and the SoC is the quantity that
+    # decides whether it may be sized. Asked one line earlier it would grade a
+    # number that governs nothing on every quiet cycle of the day, and would
+    # report "floor-limited" over a guard that was never going to discharge.
+    denied = discharge_permitted(soc, soc_floor)
+    if denied is not None:
+        # 0.0, not `wanted`, and not a negative number. Zero forbids CHARGING -
+        # one-directional, cannot cost pack energy - and forbids nothing else;
+        # a `lo` limit structurally cannot cap the discharge the loop is already
+        # doing. See the module note: this bounds THE GUARD's forced discharge,
+        # not the grid-following loop, which still has no SoC guard.
+        return PeakDecision(
+            engaged=True, floor_w=0.0, ceiling_w=ceiling,
+            ceiling_from=from_where, projected_w=projected, excess_w=ex,
+            wanted_w=wanted, floor_limited=True, soc=soc, soc_floor=soc_floor,
+            soc_floor_reason=denied,
+            reason=(
+                f"PEAK GUARD FLOOR-LIMITED: the quarter projects "
+                f"{projected:.0f} W against a {ceiling:.0f} W ceiling "
+                f"({from_where}) and {wanted:+.0f} W of discharge would hold "
+                f"it, but {denied}"
+                + ("" if soc is None else f" ({float(soc):.0f} %")
+                + ("" if soc is None or soc_floor is None
+                   else f" vs {float(soc_floor):.0f} %")
+                + ("" if soc is None else ")")
+                + " - commanding no forced discharge, so THIS QUARTER WILL GO "
+                  "OVER THE CEILING. Charging stays blocked (floor 0 W)"))
+
     floor = min(wanted, ceil_top)
     saturated = floor < wanted
     return PeakDecision(
         engaged=True, floor_w=floor, ceiling_w=ceiling, ceiling_from=from_where,
         blind=None, projected_w=projected, excess_w=ex, wanted_w=wanted,
-        saturated=saturated,
+        saturated=saturated, soc=soc, soc_floor=soc_floor,
         reason=(
             f"PEAK GUARD: quarter projects {projected:.0f} W against a "
             f"{ceiling:.0f} W ceiling ({from_where}) with "
@@ -513,5 +713,13 @@ def claim(decision: PeakDecision) -> Claim | None:
     """
     if not decision.engaged or decision.floor_w is None:
         return None
+    # Three reasons, not two. A `lo = 0 W` claim reaches the arbiter from THREE
+    # different states now - blind, floor-limited, and a genuine floor that
+    # happens to be zero - and the arbiter's explain() is what an operator reads
+    # when the battery stops charging. Spelling them the same way would make the
+    # SAFETY-24 case indistinguishable from the SAFETY-18 one on the one line
+    # that gets read.
     return Claim.limit(SOURCE, P_PEAK, lo=float(decision.floor_w),
-                       reason=("blind" if decision.blind else "quarter peak"))
+                       reason=("blind" if decision.blind
+                               else "SoC floor" if decision.floor_limited
+                               else "quarter peak"))

goodwe_controller/app/price.py

@@ -607,6 +607,26 @@ PRICE_OPTION_DEFAULTS = {
     # test_price.py section 1. A second defaults dict in the new module would be
     # a fourth place option defaults live, which is exactly the open ticket.
     "static_profile": "",
+    # OPT-04. §10.1 priority 3, and §14's `optimizer.cycle_cost_threshold_eur`.
+    #
+    # ⚠️ THEY LIVE HERE FOR CFG-03's SAKE, exactly as `static_profile` does, and
+    # FCST-03's note above `WIDEN_FACTOR` names this as the place they would
+    # land. `arbitrage_soc_floor` / `arbitrage_soc_ceiling` are the reserve
+    # arbitrage may not spend and the headroom it may not overfill; they are
+    # NOT the maintenance SoC keys, which describe a balancing cycle rather
+    # than a trading reserve and would silently couple two unrelated policies.
+    #
+    # ⚠️ AND `cycle_cost_threshold_eur` IS 0.10 BECAUSE §10.1's NAME FOR IT SAYS
+    # SO, NOT BECAUSE IT MAKES ANYTHING TRADE. On realistic Belgian day/night
+    # numbers the static spread is ~0.08 EUR/kWh, so a static-tariff site does
+    # NOT trade at this default. That is a reported result (test_arbitrage.py
+    # section 7), and whether 0.10 - or FCST-03's 2.5/1.5 - is right is one
+    # product decision that belongs to the owner. No number here was chosen to
+    # make arbitrage happen.
+    "arbitrage_enabled": False,
+    "cycle_cost_threshold_eur": 0.10,
+    "arbitrage_soc_floor": 20,
+    "arbitrage_soc_ceiling": 95,
 }
 
 # ⚠️ The keys the FRAMEWORK owns. A plugin that reads one of these is a plugin

goodwe_controller/app/validate.py

@@ -131,6 +131,25 @@ NOT READY banner already names the problem, (b) does not apply.
      installed base is on a fixed or static tariff. Refusing to start without a
      price feed would lock those sites out of the add-on entirely.
 
+  8. `peak_soc_floor` at or below `bms_soc_floor`.  [(a)]
+     SAFETY-24, and the same shape as 1, 4, 5 and 6. The peak guard can COMMAND
+     a discharge to hold a billed quarter down (SAFETY-07), and `peak_soc_floor`
+     is the EMS-side bound on how far down it may take the pack. The BMS has its
+     own cut-out and enforces it regardless. Set the EMS floor at or below that
+     cut-out and THE BMS WINS: the EMS bound can never be the thing that stops
+     the discharge, so the add-on carries an energy backstop that is incapable
+     of binding while its status page reports one - rule 5's "a disarmed guard
+     that advertises itself as armed", on the guard whose failure mode is a pack
+     that is quietly never available when it is wanted.
+
+     ⚠️ `<=`, NOT `<`, and the equality case is the interesting one. At exactly
+     equal, the EMS floor and the pack's cut-out act at the same percent, so the
+     EMS floor buys ZERO margin - it can never prevent the pack reaching the BMS
+     limit, which is the entire thing this option exists to prevent. That is not
+     "close to the rail", it is the disarmed case with an extra step. Rule 6
+     permits equality because equality there leaves one reachable point that
+     works; here it leaves none.
+
 WHAT IS DELIBERATELY *NOT* REFUSED, because it is untidy rather than unsafe:
 
   - `integrator_max_w` above `max_w`. DOCS.md is right that the only thing the
@@ -151,12 +170,18 @@ WHAT IS DELIBERATELY *NOT* REFUSED, because it is untidy rather than unsafe:
   - `maintenance_discharge_w` above `max_w`. Clamped, not contradicted, and the
     clamp names itself in the arbiter's explain() - see rule 6. Warned about,
     not refused.
+  - `peak_soc_floor` below `maintenance_soc_floor`, or above it. They are
+    DIFFERENT QUANTITIES (SAFETY-24: a drain target versus a spend limit) and
+    every ordering of the two is a legitimate site configuration. Refusing on
+    an ordering would be inventing a relationship the system does not have.
 """
 
 import logging
 
+from .arbitrage import enabled as arbitrage_on
 from .p1 import is_enabled, validate_meter_options
-from .price import PROVIDERS
+from .peakguard import PEAK_OPTION_DEFAULTS
+from .price import PRICE_OPTION_DEFAULTS, PROVIDERS
 from .static_tariff import ProfileInvalid, parse_static_profile
 
 _LOG = logging.getLogger("goodwe.config")
@@ -369,6 +394,36 @@ def validate_startup_config(opts: dict, dev_min=None, dev_max=None) -> list[str]
     # --- 7. two price sources on one site ------------------------------------
     problems.extend(_price_problems(opts))
 
+    # --- 8. an EMS energy floor the BMS does not already enforce -------------
+    # ⚠️ THE FALLBACKS ARE peakguard.PEAK_OPTION_DEFAULTS, NOT LITERALS, and
+    # that is CFG-03 read as a rule rather than as a complaint. Rule 6's own
+    # comment documents a LIVE divergence in this repo caused by typing 2000
+    # here and 2500 in main.py; this card refuses to add a second instance of
+    # it. The dict is the single in-code mirror, pinned to config.yaml by
+    # test_peakguard.py, and main.py reads the same one.
+    soc_floor = _number(opts, "peak_soc_floor",
+                        PEAK_OPTION_DEFAULTS["peak_soc_floor"])
+    bms_floor = _number(opts, "bms_soc_floor",
+                        PEAK_OPTION_DEFAULTS["bms_soc_floor"])
+    if soc_floor is None or bms_floor is None:
+        problems.append(
+            "peak_soc_floor and bms_soc_floor must both be numbers (got "
+            f"{opts.get('peak_soc_floor')!r} / {opts.get('bms_soc_floor')!r})")
+    elif soc_floor <= bms_floor:
+        problems.append(
+            f"peak_soc_floor is {soc_floor:.0f} % but the pack's own cut-out "
+            f"(bms_soc_floor) is {bms_floor:.0f} %, so SAFETY-07's forced "
+            "discharge has no EMS-side bound at all - THE BMS WINS, and it is "
+            "the pack's protection rather than yours. The guard would report a "
+            "SoC floor on its status page while being incapable of ever being "
+            "the thing that stops the discharge, and the symptom of that is a "
+            "battery repeatedly sitting at the BMS limit with no record of why "
+            f"- which looks exactly like the SoC drift the maintenance cycle "
+            f"exists to correct. Raise peak_soc_floor above {bms_floor:.0f} % "
+            f"(the shipped default is "
+            f"{PEAK_OPTION_DEFAULTS['peak_soc_floor']:.0f} %), or correct "
+            "bms_soc_floor if it does not match this pack.")
+
     # --- warnings: real, but not worth refusing over -------------------------
     integ = opts.get("integrator_max_w")
     integ_w = None if not integ else _number(opts, "integrator_max_w", None)
@@ -453,14 +508,21 @@ def _price_problems(opts: dict) -> list[str]:
     #     arbitrage verdict, so the seam is no longer inert.
     #   * main.py reads the STATIC side of that module every cycle
     #     (static_status()), which makes static_profile a real writer of `cheap`.
-    #   * main.py does NOT yet call resolve_tariff() on the control tick, and
-    #     that is deliberate: a provider fetch is network I/O and the tick is
-    #     what feeds the ESP32 heartbeat. So a configured PROVIDER still shapes
-    #     no energy decision on its own; a configured STATIC PROFILE does.
+    #   * main.py does NOT call resolve_tariff() on the control tick, and that
+    #     is still deliberate: a provider fetch is network I/O and the tick is
+    #     what feeds the ESP32 heartbeat. OPT-04 put it on a BACKGROUND REFRESH
+    #     instead (Controller.run_tariff_refresh), and the tick reads the plan
+    #     that task last stored - an attribute read, with nothing to block on.
+    #   * so a configured PROVIDER now DOES shape an energy decision, but only
+    #     through app/arbitrage.py and only while `arbitrage_enabled` is on.
+    #     With it off - the shipped configuration - the series is still fetched,
+    #     validated and published, and still moves nothing.
     #
-    # Hence two refusals below rather than one, and neither claims more than
-    # that list. When OPT-04 puts resolve_tariff() on a background refresh, the
-    # first message's middle clause is what changes - not the rule.
+    # Hence three refusals below rather than two, and none of them claims more
+    # than that list. ⚠️ THE MIDDLE CLAUSE OF THE FIRST MESSAGE IS WHAT OPT-04
+    # CHANGED, exactly as the note it replaced predicted - the RULE is
+    # untouched, because the reason it exists (main.py reads the Home Assistant
+    # pair FIRST and the other source is silent) is unchanged.
     ha_price = [k for k in ("price_now_entity", "price_avg_entity")
                 if str(opts.get(k, "") or "").strip()]
     if pid != "off" and ha_price:
@@ -469,10 +531,12 @@ def _price_problems(opts: dict) -> list[str]:
             f"{'is' if len(ha_price) == 1 else 'are'} set, which is two price "
             "sources on one site. THE HOME ASSISTANT ENTITIES WIN: they are "
             "what main.py reduces to the `cheap` flag the maintenance cycle "
-            "reads. The provider series IS consumed now - app/static_tariff.py "
-            "builds a tariff plan and an arbitrage verdict from it - but it is "
-            "not what main.py acts on, so the supplier you configured would "
-            "shape no energy decision while the status page named it. Clear "
+            "reads, and read_inputs() never looks at the other source once "
+            "they are set. The provider series is consumed - "
+            "app/static_tariff.py builds a tariff plan and an arbitrage "
+            "verdict, and app/arbitrage.py trades on it off a background "
+            "refresh - but the `cheap` flag those entities own is a separate "
+            "decision they would silently keep. Clear "
             "price_now_entity and price_avg_entity to use the "
             f"{pid!r} plugin, or set price_provider to 'off' to keep the "
             "Home Assistant sensors.")
@@ -528,6 +592,67 @@ def _price_problems(opts: dict) -> list[str]:
             "with no tariff at all rather than with a coarse one. Not refused, "
             "but filling it in is what makes that row work.", pid)
 
+    # ⚠️ OPT-04. THE THREE WAYS `arbitrage_enabled: true` CAN BE A LIE, refused
+    # rather than logged, because a strategy that is armed in the UI and cannot
+    # fire is indistinguishable from one that fires and finds no opportunity -
+    # which is the §14 warning box's case exactly. None of these can break an
+    # existing install: the option is new and ships `false`.
+    # ⚠️ CHECKED WHETHER OR NOT ARBITRAGE IS ARMED. A threshold that is not a
+    # non-negative number is a configuration error the moment it is typed, and
+    # deferring the complaint until somebody turns the strategy on would report
+    # it at the worst possible moment - the one where energy starts moving.
+    thresh = _number(opts, "cycle_cost_threshold_eur",
+                     PRICE_OPTION_DEFAULTS["cycle_cost_threshold_eur"])
+    if thresh is None or thresh < 0:
+        problems.append(
+            f"cycle_cost_threshold_eur is "
+            f"{opts.get('cycle_cost_threshold_eur')!r} and must be a "
+            f"non-negative number of EUR/kWh - it is what one round trip COSTS "
+            f"(§10.1). A negative threshold is one that every spread clears, "
+            f"including a spread of nothing, which is arbitrage with the brake "
+            f"removed rather than arbitrage tuned aggressively.")
+
+    if arbitrage_on(opts):
+        if pid == "off" and not str(opts.get("static_profile", "") or "").strip():
+            problems.append(
+                "arbitrage_enabled is on but price_provider is 'off' and "
+                "static_profile is empty, so this site has NO PRICES AT ALL. "
+                "§10.1's priority 3 is 'charge cheap, discharge expensive' and "
+                "there is nothing here to tell cheap from expensive, so the "
+                "strategy would refuse on every cycle forever while the "
+                "configuration said it was running. Configure a price_provider "
+                "or a static_profile, or turn arbitrage_enabled off.")
+        if not is_enabled(opts):
+            problems.append(
+                "arbitrage_enabled is on but meter_source is 'off', so there "
+                "is no P1 telegram stream, the capacity-tariff peak guard "
+                "never runs, and nothing measures the 15-minute average this "
+                "strategy's charging would land in. §9.1 gives peak shaving "
+                "priority over arbitrage; an unmeasured peak cannot be shaved, "
+                "so arbitrage would be deliberately creating the billed "
+                "quantity with the one guard that limits it switched off. "
+                "app/arbitrage.py refuses to trade in that state anyway - this "
+                "refuses the configuration that asks it to. Set meter_source, "
+                "or turn arbitrage_enabled off.")
+        floor = _number(opts, "arbitrage_soc_floor",
+                        PRICE_OPTION_DEFAULTS["arbitrage_soc_floor"])
+        ceiling = _number(opts, "arbitrage_soc_ceiling",
+                          PRICE_OPTION_DEFAULTS["arbitrage_soc_ceiling"])
+        if floor is None or ceiling is None:
+            problems.append(
+                f"arbitrage_soc_floor {opts.get('arbitrage_soc_floor')!r} and "
+                f"arbitrage_soc_ceiling {opts.get('arbitrage_soc_ceiling')!r} "
+                f"must both be numbers - they are percentages of pack charge.")
+        elif floor >= ceiling:
+            problems.append(
+                f"arbitrage_soc_floor is {floor:g} % and arbitrage_soc_ceiling "
+                f"is {ceiling:g} %, so there is no band left between them: "
+                f"every state of charge is either at or under the floor (no "
+                f"discharge) or at or over the ceiling (no charge), and the "
+                f"strategy can never act. The floor is the reserve it may not "
+                f"spend and the ceiling is the headroom it may not overfill, "
+                f"so the floor must be the SMALLER number.")
+
     if pid == "manual_csv" and not str(opts.get("price_csv_path", "") or "").strip():
         problems.append(
             "price_provider is 'manual_csv' but price_csv_path is empty, so "

goodwe_controller/config.yaml

@@ -163,6 +163,37 @@ options:
   peak_forecast_entity: ""
   peak_cap_w: 3500
 
+  # --- SAFETY-24: the energy backstop under SAFETY-07's forced discharge -----
+  # The peak guard can COMMAND a discharge to hold a billed quarter under
+  # peak_cap_w. Until this card it was bounded from below by nothing but the
+  # pack's own BMS cut-out - a real floor, but the PACK's protection rather than
+  # the OPERATOR's. On the 2026-08-20 replay the guard asks for 878 Wh in a day.
+  #
+  # peak_soc_floor is the percent below which that discharge is not sized. When
+  # it binds the guard reports FLOOR-LIMITED, keeps charging blocked (a 0 W
+  # floor), and the add-on raises ONE notification saying the quarter will be
+  # exceeded - failing to shave is never silent, because the euro figure is the
+  # whole point of the guard.
+  #
+  # ⚠️ THIS IS NOT maintenance_soc_floor AND MUST NOT BE MERGED WITH IT.
+  # maintenance_soc_floor is the DRAIN TARGET of the monthly recalibration: the
+  # depth the cycle deliberately goes to, once a month, on purpose. This is the
+  # depth a money-saving discharge may never reach on any day. The right values
+  # differ, and sharing one key would couple two features that want to move
+  # independently.
+  #
+  # ⚠️ 20 IS DERIVED, NOT PICKED. The reference pack is 13.8 kWh usable and the
+  # worst modelled guard day is 878 Wh, i.e. 6.4 % - so 10 points above the
+  # pack's own cut-out leaves the guard ~1.6 of its worst observed days of room
+  # and still never reaches the BMS.
+  peak_soc_floor: 20
+  # The pack's OWN cut-out percent - what the BMS enforces whatever is set
+  # above. Register 0x0560 reads 10 on the reference GW5000S-BP, consistent with
+  # DoD 90. It is here so validate.py can REFUSE a peak_soc_floor at or below
+  # it: such a floor can never bind, the BMS wins, and the status page would
+  # report an energy backstop that does not exist.
+  bms_soc_floor: 10
+
   # ⚠️ THE OLD PRICE PATH, AND IT IS NOT DEAD YET. These two entities are the
   # add-on delegating the whole price question to Home Assistant: main.py reads
   # both and derives ONE BOOLEAN, `cheap = now <= avg`, which is consumed by
@@ -240,6 +271,38 @@ options:
   # a second series: no shipped supplier publishes a dynamic injection curve.
   export_price_eur_kwh: 0.04
 
+  # --- price arbitrage (OPT-04, specs §10.1 priority 3 / §17) ---------------
+  # Charge in the cheapest published hour, discharge in the dearest, and ONLY
+  # when the spread covers what the round trip costs. Off in the shipped
+  # configuration: this is the first strategy in the add-on that moves energy
+  # against the meter on purpose, so it is armed deliberately or not at all.
+  #
+  # ⚠️ IT NEEDS A PRICE SOURCE AND A P1 METER, and validate.py refuses the
+  # combinations that cannot work rather than letting them look armed. With
+  # `price_provider: off` and no `static_profile` there is no series to trade
+  # on; with `meter_source: off` the capacity-tariff peak guard never runs, and
+  # §9.1 puts peak shaving above arbitrage - a charge nothing measures is the
+  # exact failure that guard exists to prevent.
+  #
+  # ⚠️ AT THE SHIPPED THRESHOLD A STATIC DAY/NIGHT SITE DOES NOT TRADE. The
+  # realistic Belgian day/night spread is about 0.08 EUR/kWh and the threshold
+  # below is 0.10, so the verdict is "the firm spread does not cover this
+  # cycle's cost" and nothing happens - correctly. That is a product decision
+  # (this number, and app/price.py's widening factors) and it is open with the
+  # owner; the code is correct at any value and no number was picked to make a
+  # trade appear.
+  arbitrage_enabled: false
+  # §10.1's `cycle_cost_threshold_eur`: what one full round trip costs per kWh
+  # - the wear on the pack plus the conversion losses. A spread below this is
+  # a cycle that loses money and pack life, so it is not taken.
+  cycle_cost_threshold_eur: 0.10
+  # The reserve arbitrage may not spend, and the headroom it may not overfill.
+  # Deliberately NOT the maintenance_soc_* keys: those describe a monthly
+  # balancing cycle, and pointing a trading strategy at them would couple two
+  # unrelated policies through one number.
+  arbitrage_soc_floor: 20
+  arbitrage_soc_ceiling: 95
+
   # --- site ------------------------------------------------------------------
   estop_fitted: false
   log_level: info
@@ -311,6 +374,12 @@ schema:
 
   peak_forecast_entity: str?
   peak_cap_w: int(500,15000)
+  # SAFETY-24. No "off" value: a guard that can command a discharge does not get
+  # a supported way to have no energy bound at all. The ranges OVERLAP on
+  # purpose - the schema cannot express "above bms_soc_floor", so validate.py
+  # rule 8 is what refuses the pair, with a message naming both numbers.
+  peak_soc_floor: int(5,60)
+  bms_soc_floor: int(0,50)
   price_now_entity: str?
   price_avg_entity: str?
 
@@ -337,5 +406,15 @@ schema:
   # bottoms out at 0 would make that site inexpressible.
   export_price_eur_kwh: float(-1,1)
 
+  # OPT-04. ⚠️ The threshold's lower bound is 0 and not negative: it is a COST,
+  # and a negative one is a threshold every spread clears - arbitrage with the
+  # brake removed. app/arbitrage.py refuses it in code too, because
+  # validate_startup_config() is also sim/inject.py's entry point and can be
+  # handed any dict at all.
+  arbitrage_enabled: bool
+  cycle_cost_threshold_eur: float(0,1)
+  arbitrage_soc_floor: int(5,95)
+  arbitrage_soc_ceiling: int(10,100)
+
   estop_fitted: bool
   log_level: list(trace|debug|info|warning|error)

goodwe_controller/test_arbitrage.py

@@ -0,0 +1,998 @@
+"""Runnable check for OPT-04 - price arbitrage.  `python3 test_arbitrage.py`
+
+⚠️ RUN IT WITH `python3 -B` AND PURGE `__pycache__` FIRST. TEST-15 records a
+size-preserving edit, reverted inside one second, whose OLD BYTECODE was still
+what ran - confirmed by disassembly - so a mutation graded code that was not
+there. Every mutation reported in the PR for this card was run as:
+
+    find . -name __pycache__ -type d -exec rm -rf {} + ; python3 -B test_*.py
+
+WHAT THIS FILE IS DEFENDING, in the order the card's acceptance criteria name it:
+
+  AC1  NO ACTION when the expected benefit is below `cycle_cost_threshold_eur`.
+       Measured by BISECTING the accept/refuse boundary and comparing the
+       boundary with a threshold derived here by different arithmetic - not by
+       reading a constant back out of the module that defines it.
+  AC2  NEGATIVE-PRICE HOURS NEVER TRIGGER EXPORT (§17). Both sides: the same
+       series with the sign of one hour changed must trade in one case and
+       refuse in the other, or the guard is graded by nothing.
+  AC3  YIELDS TO PEAK SHAVING (§9.1). Through the two mechanisms that already
+       exist - static_tariff's required `peak_guard_active` interlock and
+       arbiter.py's clause 2 - both driven end to end, plus the third state
+       (the guard never ran) that neither of them can express on its own.
+  AC4  NET-POSITIVE ACROSS 12 MONTHS OF REPLAYED PRICES. ⚠️ SEE SECTION 7. THIS
+       REPOSITORY HOLDS NO PRICE HISTORY AT ALL - `scenarios/` carries two days
+       of real METER data and nothing else - so the criterion as written cannot
+       be met, and it is NOT redefined here into something that can be. What
+       section 7 does instead is stated there in full, and what it cannot say is
+       stated with it.
+
+THREE RULES THIS FILE HOLDS ITSELF TO, because the reviewer will:
+
+  * EVERY VERDICT IS DERIVED. No `check("the guard works", True)`. The widened
+    threshold is bisected rather than read; the arbiter's ordering is computed
+    from the priority constants rather than typed; the replay's net figure is
+    accumulated from the same trades the strategy actually chose.
+
+  * NO STIMULUS IS EVALUATED WHERE THE QUANTITY IT GRADES IS DEGENERATE. This
+    ticket is thick with that trap and each one is asserted before it is used:
+
+      - A SERIES WHOSE CHEAPEST HOUR IS ALSO ITS FIRST HOUR cannot tell "acts in
+        the cheapest hour" from "acts in the first hour". The fixture's cheapest
+        and dearest hours are neither the first nor the last, and neither is
+        adjacent to the other.
+      - THE NEGATIVE-PRICE CASE MUST STILL CLEAR THE THRESHOLD, or §17's guard
+        is graded on a series that the spread rung already refused, and deleting
+        the guard changes nothing. The fixture's all-negative day has a
+        0.25 EUR/kWh spread, asserted before it is used.
+      - IT MUST ALSO BE THE DEAREST HOUR THAT IS NEGATIVE. A day with one
+        negative hour that is the CHEAPEST one is a day the guard never
+        consults, and it would pass with the guard removed.
+      - `peak_guard_active` FALSE AND TRUE MUST DIFFER ON THE SAME PLAN, or AC3
+        is graded by a plan that refuses for some other reason anyway.
+      - THE PEAK FLOOR MUST BE INSIDE THE ARBITRAGE CHARGE, or clause 2 clamps
+        nothing and the arbiter check passes for an unbound claim.
+      - SoC MUST START STRICTLY BETWEEN FLOOR AND CEILING in every check that is
+        not about SoC, or a refusal attributed to price is really about charge.
+
+  * A FAILED READ IS NOT AN OBSERVATION. No plan yet resolved, a plan that has
+    aged out, an hour the series does not cover, an unread state of charge and a
+    peak guard that never ran are five DIFFERENT reasons - and none of them is
+    an answer. Section 5 drives all five and asserts they are distinguishable.
+"""
+
+import ast
+import inspect
+import os
+import sys
+from datetime import datetime, timedelta, timezone
+
+from app import arbitrage as _arb
+from app import main as _main
+from app import peakguard as _pg
+from app.arbiter import (
+    P_ARBITRAGE, P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_PEAK, P_SAFETY, Claim,
+    resolve,
+)
+from app.arbitrage import (
+    CHARGE, DISCHARGE, ArbitrageDecision, claim, decide, enabled, soc_bounds,
+    threshold_eur_kwh,
+)
+from app.price import (
+    PRICE_OPTION_DEFAULTS, Absence, PricePoint, PriceProduct, PriceRead,
+    Uncertainty, WIDEN_FACTOR, brussels, required_spread, widen_factor,
+)
+from app.static_tariff import TariffPlan, parse_static_profile, static_plan
+from app.validate import validate_startup_config
+
+UTC = timezone.utc
+HERE = os.path.dirname(os.path.abspath(__file__))
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+def raises(fn, exc):
+    """True only if fn() raises EXACTLY `exc`. A different exception is False -
+    'it was refused' and 'it broke on the way to being refused' must not grade
+    the same."""
+    try:
+        fn()
+    except exc:
+        return True
+    except Exception:
+        return False
+    return False
+
+
+class _Raised:
+    def __init__(self, e):
+        self.e = e
+
+    def __eq__(self, other):
+        return False
+
+    def __hash__(self):
+        return id(self)
+
+    def __bool__(self):
+        return False
+
+    def __repr__(self):
+        return f"<raised {type(self.e).__name__}: {self.e}>"
+
+
+def derived(fn):
+    """A value a check grades. An exception becomes `_Raised`, which equals
+    nothing and is falsy - NOT a default. See test_price.py for why."""
+    try:
+        return fn()
+    except Exception as e:                                    # noqa: BLE001
+        return _Raised(e)
+
+
+def _fn(module_src, name):
+    tree = ast.parse(module_src)
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) \
+                and node.name == name:
+            return node
+    return None
+
+
+def _awaited(node):
+    """Every name awaited anywhere inside `node`."""
+    out = set()
+    for n in ast.walk(node):
+        if not isinstance(n, ast.Await):
+            continue
+        f = n.value.func if isinstance(n.value, ast.Call) else n.value
+        out.add(getattr(f, "id", None) or getattr(f, "attr", None))
+    return {n for n in out if n}
+
+
+def near(a, b, tol=1e-9):
+    try:
+        return abs(float(a) - float(b)) <= tol
+    except (TypeError, ValueError):
+        return False
+
+
+# --------------------------------------------------------------------------- #
+# Fixtures
+# --------------------------------------------------------------------------- #
+
+D0 = datetime(2026, 1, 15, 0, 0, tzinfo=UTC)
+MAX_W = 2000.0
+NO_WEATHER = frozenset({Uncertainty.NO_WEATHER})
+
+
+def plan_of(prices, *, product=PriceProduct.DAY_AHEAD, start=D0,
+            provisional=False, provider="generic_dayahead"):
+    """A TariffPlan over consecutive hourly prices starting at `start`.
+
+    `provisional` may be a single bool or a per-hour sequence.
+    """
+    flags = ([bool(provisional)] * len(prices)
+             if isinstance(provisional, bool) else list(provisional))
+    pts = [PricePoint(start=brussels(start + timedelta(hours=i)),
+                      eur_per_kwh=round(float(p), 6), provisional=f)
+           for i, (p, f) in enumerate(zip(prices, flags))]
+    return TariffPlan(PriceRead.series(pts, product, provider, start), None,
+                      "provider")
+
+
+def hour(i, *, start=D0):
+    """Mid-hour instant of slot `i` - never a boundary, where two answers agree."""
+    return start + timedelta(hours=i, minutes=30)
+
+
+def act(plan, i, *, peak=False, soc=50.0, thresh=0.0, unc=frozenset(),
+        floor=20.0, ceiling=95.0, start=D0):
+    """`decide()` with the fixture's constants filled in. One call shape."""
+    return decide(plan, hour(i, start=start), peak_guard_active=peak, soc=soc,
+                  min_spread_eur_kwh=thresh, uncertainty=unc,
+                  charge_w=MAX_W, discharge_w=MAX_W,
+                  soc_floor_pct=floor, soc_ceiling_pct=ceiling)
+
+
+# ⚠️ THE SHAPE OF THIS DAY IS THE WHOLE DEGENERACY ARGUMENT. The cheapest hour
+# is slot 3 and the dearest is slot 18: neither is first, neither is last,
+# neither is adjacent to the other, and the last hour is not the dearest. An
+# implementation that acted on "the first hour", "the last hour", "the hour
+# after the cheapest" or "min/max swapped" produces a different answer from the
+# correct one at every probe below.
+DAY = [0.240, 0.230, 0.215, 0.150, 0.190, 0.225,
+       0.260, 0.300, 0.320, 0.310, 0.295, 0.280,
+       0.270, 0.275, 0.285, 0.300, 0.330, 0.360,
+       0.410, 0.380, 0.340, 0.300, 0.270, 0.250]
+CHEAP_H, DEAR_H = 3, 18
+SPREAD = round(max(DAY) - min(DAY), 6)
+
+print("0. the fixture is not degenerate (asserted before anything is graded)")
+check(f"the cheapest hour is slot {CHEAP_H} and the dearest slot {DEAR_H} - "
+      f"neither first, neither last, and not adjacent",
+      DAY.index(min(DAY)) == CHEAP_H and DAY.index(max(DAY)) == DEAR_H
+      and CHEAP_H not in (0, len(DAY) - 1) and DEAR_H not in (0, len(DAY) - 1)
+      and abs(CHEAP_H - DEAR_H) > 1)
+check(f"...and its spread is {SPREAD:.4f} EUR/kWh, comfortably above the "
+      f"shipped threshold, so the checks below grade the strategy rather than "
+      f"the spread rung",
+      SPREAD > PRICE_OPTION_DEFAULTS["cycle_cost_threshold_eur"] * 1.5)
+check("...and its minimum and its maximum each occur EXACTLY ONCE, so 'the "
+      "cheapest hour' names one hour and an undefined tie-break is not what "
+      "any check below is really grading",
+      DAY.count(min(DAY)) == 1 and DAY.count(max(DAY)) == 1)
+check("...and no price in it is zero or negative, so section 2's sign flip is "
+      "the only source of a negative price in this file",
+      all(p > 0 for p in DAY))
+
+
+# --------------------------------------------------------------------------- #
+print()
+print("1. AC1 - nothing happens below cycle_cost_threshold_eur")
+
+PLAN = plan_of(DAY)
+
+_buy = act(PLAN, CHEAP_H, thresh=0.0)
+_sell = act(PLAN, DEAR_H, thresh=0.0)
+check("with a zero threshold the cheapest hour CHARGES and the dearest hour "
+      "DISCHARGES - the two actions this strategy has, both reachable",
+      _buy.action == CHARGE and _sell.action == DISCHARGE)
+check("...charging is a NEGATIVE setpoint and discharging a positive one, "
+      "which is the sign convention maintenance's `hi <= 0` shaping and "
+      "peakguard's `lo >= 0` floor are both written against",
+      _buy.power_w is not None and _sell.power_w is not None
+      and _buy.power_w < 0 < _sell.power_w)
+check("...and an hour that is neither buy nor sell does NOTHING, so the "
+      "strategy is not simply always-on",
+      act(PLAN, 10, thresh=0.0).action is None)
+
+# ⚠️ MEASURED, NOT READ. The boundary is found by bisection on the threshold,
+# so this grades the shipped refusal ladder rather than restating a constant
+# that lives in the module under test.
+def boundary(unc, plan=PLAN, lo=0.0, hi=10.0):
+    """The largest threshold this plan still trades under, to 1e-9."""
+    for _ in range(200):
+        mid = (lo + hi) / 2.0
+        if act(plan, CHEAP_H, thresh=mid, unc=unc).action is None:
+            hi = mid
+        else:
+            lo = mid
+    return lo
+
+
+_b_plain = derived(lambda: boundary(frozenset()))
+check(f"the accept/refuse boundary with no uncertainty is the spread itself "
+      f"({_b_plain!r} vs {SPREAD}) - a cycle trades exactly while its spread "
+      f"covers its cost, and not one cent further",
+      near(_b_plain, SPREAD, 1e-6))
+check("...so a threshold a hair ABOVE the spread refuses and a hair BELOW it "
+      "trades: the boundary is a boundary, not a slope",
+      act(PLAN, CHEAP_H, thresh=SPREAD + 1e-4).action is None
+      and act(PLAN, CHEAP_H, thresh=SPREAD - 1e-4).action == CHARGE)
+_refusal = act(PLAN, CHEAP_H, thresh=SPREAD + 1e-4).reason
+check("...and the refusal SAYS the spread and the number it had to cover, so "
+      "an operator can tell 'no opportunity' from 'broken'",
+      "does not cover" in _refusal or "needs" in _refusal)
+
+# The widening path, end to end: the boundary must move by exactly the declared
+# factor. Derived twice - once by bisection, once through required_spread().
+_b_wide = derived(lambda: boundary(NO_WEATHER))
+check("the two widening factors this repository ships DIFFER, or 'max', 'sum' "
+      "and 'product' would be indistinguishable combination rules",
+      len(set(WIDEN_FACTOR.values())) == len(WIDEN_FACTOR))
+check(f"a widened threshold moves the boundary by exactly the declared factor: "
+      f"{_b_wide!r} x {widen_factor(NO_WEATHER)} = {_b_plain!r}",
+      near(_b_wide * widen_factor(NO_WEATHER), _b_plain, 1e-6))
+check("...which is the same number required_spread() computes from the other "
+      "direction, so the widening is one mechanism and not two",
+      isinstance(_b_wide, float)
+      and near(required_spread(_b_wide, NO_WEATHER), _b_plain, 1e-6))
+_wide_at = round(SPREAD / widen_factor(NO_WEATHER) + 1e-4, 6)
+check("...and the difference is LIVE, not arithmetic on paper: at one "
+      "threshold the plain site trades and the no-weather site does not",
+      act(PLAN, CHEAP_H, thresh=_wide_at, unc=frozenset()).action == CHARGE
+      and act(PLAN, CHEAP_H, thresh=_wide_at, unc=NO_WEATHER).action is None)
+
+# The threshold this add-on SHIPS, against the tariff most of the installed
+# base is on. Reported, not tuned - see the module docstring in app/arbitrage.py.
+STATIC_BE = ('{"periods": ['
+             '{"name":"day","from":"07:00","to":"22:00","days":"all",'
+             '"eur_per_kwh":0.34},'
+             '{"name":"night","from":"22:00","to":"07:00","days":"all",'
+             '"eur_per_kwh":0.26}]}')
+_st_plan = derived(lambda: static_plan({"static_profile": STATIC_BE}, D0,
+                                       D0 + timedelta(days=2), now=D0))
+_st_spread = derived(lambda: _st_plan.spread_eur_kwh())
+check(f"a realistic Belgian day/night profile has a spread of "
+      f"{_st_spread!r} EUR/kWh - the number the product decision is about",
+      near(_st_spread, 0.08))
+_shipped = PRICE_OPTION_DEFAULTS["cycle_cost_threshold_eur"]
+_st_verdict = derived(lambda: _st_plan.arbitrage(
+    peak_guard_active=False, min_spread_eur_kwh=_shipped,
+    uncertainty=NO_WEATHER))
+check(f"...and at the SHIPPED threshold of {_shipped} EUR/kWh that site does "
+      f"NOT trade. Reported, not tuned: the shipped default is the owner's "
+      f"decision and this check exists to make the consequence visible rather "
+      f"than to argue with it",
+      _st_verdict.kind == "refused")
+check(f"...and it is the SPREAD rung that refuses it, not a flat-tariff or a "
+      f"provisional rung that would have refused any static site whatever the "
+      f"threshold was: {_st_spread!r} < "
+      f"{required_spread(_shipped, NO_WEATHER)!r}",
+      _st_spread < required_spread(_shipped, NO_WEATHER)
+      and "cover its own round-trip cost" in _st_verdict.reason)
+check("...while the SAME site trades once the threshold is set below its "
+      "spread, so the refusal above is about the number and not about static "
+      "tariffs being unsupported",
+      derived(lambda: _st_plan.arbitrage(
+          peak_guard_active=False,
+          min_spread_eur_kwh=_st_spread / widen_factor(NO_WEATHER) - 1e-4,
+          uncertainty=NO_WEATHER).kind) == "allowed")
+
+
+# --------------------------------------------------------------------------- #
+print()
+print("2. AC2 - §17: a negative-price hour never triggers export")
+
+# ⚠️ THE WHOLE DAY IS NEGATIVE AND THE DEAREST HOUR IS THE LEAST NEGATIVE ONE.
+# That is the case a guard written as "the spread is positive" or "not all
+# prices are negative" lets through, and it is the realistic one: Belgian
+# day-ahead has gone negative for a whole summer midday block.
+NEG = [round(-0.30 + 0.010 * i, 6) for i in range(24)]     # -0.30 .. -0.07
+NEG[DEAR_H] = -0.05                                        # dearest, still < 0
+NEG_PLAN = plan_of(NEG)
+_neg_spread = round(max(NEG) - min(NEG), 6)
+check(f"the negative day's dearest hour is slot {DEAR_H} and it is BELOW ZERO "
+      f"({max(NEG):+.4f}), which is the only shape that exercises the guard",
+      NEG.index(max(NEG)) == DEAR_H and max(NEG) < 0)
+check(f"...and its spread is {_neg_spread:.4f} EUR/kWh, which CLEARS the "
+      f"shipped threshold - so the trade is refused by §17 and not by a rung "
+      f"that would have refused it anyway",
+      _neg_spread > required_spread(_shipped, NO_WEATHER))
+_neg_sell = act(NEG_PLAN, DEAR_H, thresh=_shipped, unc=NO_WEATHER)
+check("selling into a negative-price hour is REFUSED", _neg_sell.action is None)
+check("...and the reason names §17 and the price, so the log says which rule "
+      "and not merely that nothing happened",
+      "§17" in _neg_sell.reason and "-0.0500" in _neg_sell.reason)
+# The other side of the guard, without which nothing above is discriminating.
+POS = list(NEG)
+POS[DEAR_H] = +0.05          # the SAME day with ONE SIGN FLIPPED
+POS_PLAN = plan_of(POS)
+_pos_sell = act(POS_PLAN, DEAR_H, thresh=_shipped, unc=NO_WEATHER)
+check("...while the SAME series with that one price's SIGN FLIPPED does "
+      "discharge - so the refusal is about the sign and about nothing else "
+      "that differs between the two runs",
+      _pos_sell.action == DISCHARGE)
+check("...and the two runs differ in exactly one price, by exactly its sign",
+      sum(1 for a, b in zip(NEG, POS) if a != b) == 1
+      and near(NEG[DEAR_H], -POS[DEAR_H]))
+# §17 is about EXPORT. Buying in a negative hour is the right thing to do.
+check("BUYING in a negative-price hour is still permitted - §17 forbids "
+      "exporting into one, and a guard that also blocked the buy would be "
+      "refusing to be paid to charge",
+      act(NEG_PLAN, NEG.index(min(NEG)), thresh=_shipped,
+          unc=NO_WEATHER).action == CHARGE)
+# A zero price is the boundary and it is NOT negative.
+ZERO = list(POS)
+ZERO[DEAR_H] = 0.0
+check("a sell hour priced at exactly ZERO is not a negative one, so the guard "
+      "is `< 0` rather than `<= 0` - and the boundary is stated by a check "
+      "rather than left to whoever reads the operator",
+      act(plan_of(ZERO), DEAR_H, thresh=_shipped, unc=NO_WEATHER).action
+      == DISCHARGE)
+
+
+# --------------------------------------------------------------------------- #
+print()
+print("3. AC3 - peak shaving outranks arbitrage, through both mechanisms")
+
+# MECHANISM 1: static_tariff's interlock. The spread is not obtainable without
+# answering the question, so this cannot be skipped by a caller.
+_clear = act(PLAN, CHEAP_H, thresh=_shipped, unc=NO_WEATHER, peak=False)
+_held = act(PLAN, CHEAP_H, thresh=_shipped, unc=NO_WEATHER, peak=True)
+check("on ONE plan, at ONE instant, with ONE threshold, the peak guard being "
+      "CLEAR trades and the peak guard HOLDING does not - the only difference "
+      "between the two runs is that boolean",
+      _clear.action == CHARGE and _held.action is None)
+check("...and the refusal cites §10.1's ordering and the capaciteitstarief, so "
+      "it cannot be mistaken for 'no opportunity today'",
+      "§10.1" in _held.reason and "peak" in _held.reason.lower())
+check("...and the interlock is REQUIRED rather than defaulted: "
+      "TariffPlan.arbitrage() has no default for peak_guard_active, so a "
+      "caller cannot reach a spread without having answered it",
+      all(p.default is inspect.Parameter.empty
+          for n, p in inspect.signature(TariffPlan.arbitrage).parameters.items()
+          if n != "self"))
+check("...nor does arbitrage.decide(): every argument of it is keyword-only "
+      "with no default, for the same reason",
+      all(p.default is inspect.Parameter.empty
+          and (p.kind is inspect.Parameter.KEYWORD_ONLY or n in ("plan", "now"))
+          for n, p in inspect.signature(decide).parameters.items()))
+
+# MECHANISM 2: the arbiter. Even with the interlock bypassed, the peak claim
+# binds the arbitrage claim - because P_PEAK > P_ARBITRAGE and clause 2 applies
+# every limit at or above the winner's priority.
+check("the priority ordering IS §10.1's, derived from the constants rather "
+      "than asserted about them: safety > maintenance > peak > maintenance "
+      "shaping > arbitrage > loop",
+      P_SAFETY > P_MAINTENANCE > P_PEAK > P_MAINT_SHAPE > P_ARBITRAGE > P_LOOP)
+_charge_claim = claim(_clear)
+PEAK_FLOOR = 500.0
+check(f"the peak floor used below ({PEAK_FLOOR:+.0f} W) is strictly INSIDE the "
+      f"arbitrage charge ({_charge_claim.value:+.0f} W), or clause 2 would "
+      f"clamp nothing and the resolution would pass for an unbound claim",
+      _charge_claim.value < PEAK_FLOOR)
+_res = resolve([
+    Claim.limit("supervised", P_SAFETY, -MAX_W, MAX_W, "max_w"),
+    Claim.limit(_pg.SOURCE, P_PEAK, lo=PEAK_FLOOR, reason="quarter peak"),
+    _charge_claim,
+    Claim.set("loop", P_LOOP, -120.0, "following"),
+])
+check(f"an arbitrage charge that WINS the arbitration is still clamped by the "
+      f"peak floor: {_res.explain()}",
+      near(_res.target_w, PEAK_FLOOR) and _res.winner == _arb.SOURCE
+      and _res.bound_by is not None and _pg.SOURCE in _res.bound_by)
+check("...and it is P_PEAK > P_ARBITRAGE that does it, not luck: the identical "
+      "floor filed BELOW the arbitrage priority does not bind at all",
+      near(resolve([
+          Claim.limit("supervised", P_SAFETY, -MAX_W, MAX_W, "max_w"),
+          Claim.limit(_pg.SOURCE, P_ARBITRAGE - 1, lo=PEAK_FLOOR),
+          _charge_claim]).target_w, _charge_claim.value))
+# The floor really is peakguard's own claim, not a hand-rolled stand-in.
+_pd = _pg.PeakDecision(engaged=True, floor_w=PEAK_FLOOR, ceiling_w=3500.0,
+                       ceiling_from=_pg.CEILING_CAP, reason="test")
+_real_peak = derived(lambda: _pg.claim(_pd))
+check("...and the claim doing the clamping is peakguard.claim()'s own output, "
+      "so this grades the shipped floor rather than a transcription of it",
+      _real_peak is not None and _real_peak.priority == P_PEAK
+      and near(resolve([
+          Claim.limit("supervised", P_SAFETY, -MAX_W, MAX_W, "max_w"),
+          _real_peak, _charge_claim]).target_w, PEAK_FLOOR))
+# ...and a DISCHARGE is deliberately not capped by it (§9.1: only offtake bills).
+check("a peak floor does NOT cap an arbitrage DISCHARGE - §9.1 bills offtake "
+      "only, and peakguard's claim leaves `hi` at +inf to say so",
+      near(resolve([
+          Claim.limit("supervised", P_SAFETY, -MAX_W, MAX_W, "max_w"),
+          _real_peak, claim(_sell)]).target_w, MAX_W))
+
+# MECHANISM 3, which is neither of the above: the guard that never ran.
+_blind = decide(PLAN, hour(CHEAP_H), peak_guard_active=None, soc=50.0,
+                min_spread_eur_kwh=_shipped, uncertainty=NO_WEATHER,
+                charge_w=MAX_W, discharge_w=MAX_W,
+                soc_floor_pct=20.0, soc_ceiling_pct=95.0)
+check("a peak guard that NEVER RAN is a third state, and it refuses: `None` is "
+      "not `False`, and 'nothing measured the quarter' is not 'the quarter is "
+      "clear'",
+      _blind.action is None and "NOT COVERED" in _blind.reason)
+check("...and it is distinguishable from the guard HOLDING, so the status page "
+      "does not report an unmeasured install as a peak event",
+      _blind.reason != _held.reason)
+# ⚠️ AND THE THREE STATES ARE GRADED ON THE SHIPPED EXPRESSION. `guard_state()`
+# is what main.py's cycle() passes; a transcription of it here would be the
+# defect `shaping_suspended()` was extracted to remove - a mutation back to a
+# bare `engaged_now` left every suite green in the case that record describes.
+_idle_pd = _pg.PeakDecision(engaged=False, floor_w=None, ceiling_w=3500.0,
+                            ceiling_from=_pg.CEILING_CAP, reason="idle")
+check("guard_state() maps peakguard's three real situations onto the three "
+      "states decide() takes, and None is NOT False",
+      _arb.guard_state(None) is None
+      and _arb.guard_state(_idle_pd) is False
+      and _arb.guard_state(_pd) is True)
+check("...and cycle() calls it rather than inlining the conjunction, so a "
+      "mutation back to a bare `engaged_now` is red rather than silent",
+      "arbitrage.guard_state(self.peak_guard)" in
+      ast.get_source_segment(
+          open(os.path.join(HERE, "app", "main.py"), encoding="utf-8").read(),
+          _fn(open(os.path.join(HERE, "app", "main.py"),
+                   encoding="utf-8").read(), "cycle")))
+
+
+# --------------------------------------------------------------------------- #
+print()
+print("4. §8.3 - no irreversible action on a provisional price")
+
+# The realistic shape: the feed has published the morning and the afternoon is
+# carried from yesterday. The CHEAPEST hour of the whole day is in the carried
+# part, which is the lure - and the firm part still has a tradeable spread, so
+# a correct implementation trades on the firm hours rather than doing nothing.
+_flags = [i >= 12 for i in range(24)]
+_part = plan_of(DAY, provisional=_flags)
+_firm = [p for p, f in zip(DAY, _flags) if not f]
+check("the fixture's carried half contains the day's DEAREST hour, so 'the "
+      "carried hours are ignored' is graded by a case where ignoring them "
+      "changes the answer",
+      _flags[DEAR_H] is True)
+check("...and the firm half still has a spread over the shipped threshold, so "
+      "the correct behaviour is to TRADE on the firm hours - a check that "
+      "expected silence would pass for an implementation that gave up",
+      round(max(_firm) - min(_firm), 6) > required_spread(_shipped, NO_WEATHER))
+_p_buy = act(_part, CHEAP_H, thresh=_shipped, unc=NO_WEATHER)
+_p_sell_carried = act(_part, DEAR_H, thresh=_shipped, unc=NO_WEATHER)
+_p_sell_firm = act(_part, _firm.index(max(_firm)), thresh=_shipped,
+                   unc=NO_WEATHER)
+check("the firm cheapest hour still CHARGES", _p_buy.action == CHARGE)
+check("...the firm dearest hour still DISCHARGES", _p_sell_firm.action == DISCHARGE)
+check("...and the CARRIED hour that is the day's real maximum does nothing - "
+      "it is not the sell hour, because it is not an observation",
+      _p_sell_carried.action is None)
+_all_prov = act(plan_of(DAY, provisional=True), CHEAP_H, thresh=0.0)
+check("a FULLY carried series trades nothing at all, whatever its spread - "
+      "§11.4's fallback plan and a pre-publication day both land here",
+      _all_prov.action is None and "§8.3" in _all_prov.reason)
+# ⚠️ A SURVIVOR, REPORTED RATHER THAN HIDDEN, AND THE ARGUMENT THAT MAKES IT
+# HARMLESS. Disabling `_act`'s own `if here.provisional:` guard changes no
+# answer in this file, and it cannot: the verdict's `cheapest`/`dearest` are
+# firm by `_firm_or_die`, starts are unique within a series, and the point
+# `at()` returns for a matched instant is therefore THE SAME OBJECT. The guard
+# is unreachable by construction, exactly the way peakguard's seventh mutation
+# is - and the honest response is to demonstrate the construction rather than
+# to write a check that cannot bite. It stays because it is the only thing
+# standing if a future caller ever hands `_act` a plan the verdict did not come
+# from.
+_probe = _arb._point_at(PLAN, hour(CHEAP_H))
+_verdict = PLAN.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.0,
+                          uncertainty=frozenset())
+check("the point matched for the buy hour IS the verdict's own cheapest point, "
+      "by identity - which is why `_act`'s provisional guard is unreachable "
+      "here, and why disabling it changes nothing this file can observe",
+      _probe is _verdict.cheapest and _probe.provisional is False)
+check("...and the starts really are unique in a series, which is the other "
+      "half of that argument",
+      len({p.start for p in PLAN.read.points}) == len(PLAN.read.points))
+
+
+# --------------------------------------------------------------------------- #
+print()
+print("4b. §10.2 hysteresis - a price on the boundary cannot flap at 30 s")
+
+# The card's technical notes ask for this to be "testable by replaying a price
+# series oscillating around the threshold and asserting a bounded transition
+# count". ⚠️ THE OSCILLATION IS THE POINT: these prices cross the shipped
+# threshold repeatedly, hour to hour, which is the shape that makes a
+# per-tick price-vs-threshold comparison flap.
+FLAP = [0.20 + (0.101 if i % 2 else 0.0) for i in range(24)]
+_flap_plan = plan_of(FLAP)
+check("the flapping fixture really does straddle the threshold hour by hour - "
+      "every neighbouring pair differs by more than it",
+      all(abs(FLAP[i] - FLAP[i + 1]) > _shipped for i in range(len(FLAP) - 1)))
+_ticks = []
+for _t in range(24 * 120):                 # 24 h at the 30 s control cadence
+    _d = decide(_flap_plan, D0 + timedelta(seconds=30 * _t),
+                peak_guard_active=False, soc=50.0, min_spread_eur_kwh=_shipped,
+                uncertainty=frozenset(), charge_w=MAX_W, discharge_w=MAX_W,
+                soc_floor_pct=20.0, soc_ceiling_pct=95.0)
+    _ticks.append(_d.action)
+_transitions = sum(1 for a, b in zip(_ticks, _ticks[1:]) if a != b)
+check(f"over {len(_ticks)} ticks of a day whose price crosses the threshold "
+      f"every hour, the strategy changes what it is doing {_transitions} "
+      f"times - not once per tick, because the decision is 'which HOUR is "
+      f"this', and an hour is 3600 s wide",
+      _transitions <= 4)
+check("...and it is not zero, so the bound above is a bound on a strategy that "
+      "acted rather than on one that did nothing",
+      _transitions > 0 and CHARGE in _ticks and DISCHARGE in _ticks)
+
+
+# --------------------------------------------------------------------------- #
+print()
+print("5. a failed read is not an observation - five reasons, all different")
+
+_no_plan = act(None, CHEAP_H, thresh=_shipped)
+_no_soc = act(PLAN, CHEAP_H, thresh=_shipped, soc=None)
+_uncovered = decide(PLAN, D0 + timedelta(days=9), peak_guard_active=False,
+                    soc=50.0, min_spread_eur_kwh=_shipped,
+                    uncertainty=frozenset(), charge_w=MAX_W, discharge_w=MAX_W,
+                    soc_floor_pct=20.0, soc_ceiling_pct=95.0)
+_reasons = [_no_plan.reason, _no_soc.reason, _uncovered.reason, _blind.reason,
+            _held.reason]
+check("no plan, no state of charge, an uncovered hour, an unmeasured peak and "
+      "a held peak are five DISTINCT reasons - none of them silence, none of "
+      "them each other",
+      len(set(_reasons)) == 5
+      and all(a.action is None for a in
+              (_no_plan, _no_soc, _uncovered, _blind, _held)))
+check("...and none of them is empty: `_idle` refuses a blank reason at the "
+      "call site, so a status line cannot decay into nothing",
+      all(r.strip() for r in _reasons)
+      and raises(lambda: _arb._idle(""), ValueError))
+# The no-data plan: `price_at` must RAISE rather than answer 0.0, and 0.0 is
+# exactly the value that would make every hour look like the cheapest.
+_empty = TariffPlan(PriceRead.no_data("nothing was fetched"), None, "none")
+check("a plan built on a no_data read trades nothing, and asking it for a "
+      "price RAISES rather than answering 0.0 - 'no data' becoming 'free' is "
+      "the defect this whole type family exists to remove",
+      act(_empty, 0, thresh=0.0).action is None
+      and raises(lambda: _empty.price_at(hour(0)), Absence))
+
+# SoC: the reserve and the headroom, each graded where the other cannot explain
+# the answer.
+check("at the reserve FLOOR the sell hour does nothing, and the reason names "
+      "the floor rather than the price",
+      act(PLAN, DEAR_H, thresh=0.0, soc=20.0, floor=20.0).action is None
+      and "reserve floor" in act(PLAN, DEAR_H, thresh=0.0, soc=20.0,
+                                 floor=20.0).reason)
+check("...one percent above it, the same hour sells - so the refusal is about "
+      "the floor and not about the hour",
+      act(PLAN, DEAR_H, thresh=0.0, soc=21.0, floor=20.0).action == DISCHARGE)
+check("at the CEILING the buy hour does nothing, and one percent below it buys",
+      act(PLAN, CHEAP_H, thresh=0.0, soc=95.0, ceiling=95.0).action is None
+      and act(PLAN, CHEAP_H, thresh=0.0, soc=94.0,
+              ceiling=95.0).action == CHARGE)
+check("...and a FULL pack in a NEGATIVE sell hour reports §17 rather than the "
+      "SoC - the rule that actually forbids the trade is the one logged",
+      "§17" in act(NEG_PLAN, DEAR_H, thresh=0.0, soc=20.0,
+                   floor=20.0).reason)
+
+
+# --------------------------------------------------------------------------- #
+print()
+print("6. the claim, and what NO claim means")
+
+check("an idle decision files NO claim - not a 0 W one, which would out-rank "
+      "the loop with a park order and silently switch self-consumption off on "
+      "every site that armed arbitrage",
+      claim(_held) is None and claim(_no_plan) is None)
+_idle_res = resolve([Claim.limit("supervised", P_SAFETY, -MAX_W, MAX_W),
+                     Claim.set("loop", P_LOOP, -120.0, "following")])
+check("...demonstrated through the arbiter: with arbitrage idle the loop's "
+      "number is what goes on the wire, unchanged",
+      near(_idle_res.target_w, -120.0) and _idle_res.winner == "loop")
+check("...whereas a trading decision DOES out-rank the loop, which is the "
+      "point: the loop would otherwise cancel every charge the moment it "
+      "created import",
+      resolve([Claim.limit("supervised", P_SAFETY, -MAX_W, MAX_W),
+               Claim.set("loop", P_LOOP, -120.0, "following"),
+               _charge_claim]).winner == _arb.SOURCE)
+check("the claim carries the §10.2 reason code, so the action log names the "
+      "strategy rather than describing it",
+      _charge_claim.reason == CHARGE and claim(_sell).reason == DISCHARGE)
+# SAFETY-02's arming gate, which an arbitrage winner must NOT leave armed.
+check("SAFETY-02 is disarmed while arbitrage drives, exactly as it is during a "
+      "maintenance grid charge: the strategy commands AGAINST the meter on "
+      "purpose, so sign agreement is not the expectation",
+      _main.loop_is_following(
+          resolve([Claim.limit("supervised", P_SAFETY, -MAX_W, MAX_W),
+                   Claim.set("loop", P_LOOP, -120.0, "following"),
+                   _charge_claim]), -120.0) is False)
+check("...and it is still ARMED when arbitrage is idle and the loop wins, so "
+      "the line above is not simply a watchdog that never fires",
+      _main.loop_is_following(_idle_res, -120.0) is True)
+
+
+# --------------------------------------------------------------------------- #
+print()
+print("7. AC4 - 12 months of replayed prices, and what could NOT be measured")
+
+# ⚠️ READ THIS BEFORE READING THE NUMBERS BELOW.
+#
+# The acceptance criterion is "net-positive across 12 months of replayed prices
+# in simulation". THIS REPOSITORY CONTAINS NO PRICE HISTORY. `scenarios/` holds
+# three gzipped captures and all three are P1 METER data - two days of net power
+# from the reference house plus one age capture. There is no EPEX archive, no
+# supplier export, and no CSV of a past year anywhere in the tree.
+#
+# TARIFF-01 and SAFETY-07 both hit an unmeetable criterion and both DECLINED to
+# redefine it into one they could pass, and both were upheld for it. So:
+#
+#   AC4 IS NOT MET, AND IT IS NOT RESTATED. What follows is not a replay of the
+#   Belgian market; it is a replay of a SYNTHETIC year, and a euro figure from a
+#   synthetic year grades the generator that produced it, not the strategy.
+#
+# What the replay is FOR is the half of AC4 that does not need real prices. Net
+# positivity here is not an empirical outcome to be hoped for - it is a property
+# of the decision rule, and the replay's job is to demonstrate that the property
+# holds on every trade the strategy actually chose, over a year of varied days
+# including negative-price ones. Stated as an invariant:
+#
+#     every executed round trip has (sell - buy) >= required >= threshold,
+#     so its benefit net of the threshold is >= 0, on every day, always.
+#
+# That is asserted per trade below, over 365 days. What it does NOT establish -
+# and no synthetic series could - is how OFTEN a real Belgian year clears the
+# threshold, which is the actual open question and the reason the shipped
+# default is a product decision rather than a number this card picked.
+
+def synth_day(n):
+    """A synthetic all-in day-ahead day. Deterministic, seeded by day number.
+
+    A seasonal base plus a two-peak diurnal shape plus a per-day amplitude, and
+    a summer midday block that is driven NEGATIVE - the §17 case has to occur in
+    the replay or the replay never exercises it.
+    """
+    import math
+    base = 0.22 + 0.06 * math.cos(2 * math.pi * (n - 10) / 365.0)
+    amp = 0.05 + 0.09 * abs(math.sin(n * 1.7))          # varies day to day
+    out = []
+    for h in range(24):
+        shape = (math.sin(2 * math.pi * (h - 8) / 24.0)
+                 + 0.6 * math.sin(4 * math.pi * (h - 5) / 24.0))
+        p = base + amp * shape
+        if 140 < n < 230 and 11 <= h <= 14:             # summer solar glut
+            p -= 0.45 * amp * 6
+        out.append(round(p, 6))
+    return out
+
+
+REPLAY_START = datetime(2026, 1, 1, 0, 0, tzinfo=UTC)
+CAP_WH = 10000.0            # the pack the replay moves energy through
+STEP_PCT = 100.0 * MAX_W / CAP_WH    # one hour at the rail, in percent
+
+
+def replay(threshold, unc, *, soc0=50.0, floor=20.0, ceiling=95.0):
+    """Run `decide()` over a synthetic year. Returns the ledger, nothing else."""
+    soc = soc0
+    trades, holdings, bad = [], [], []
+    neg_sells = 0
+    for n in range(365):
+        day0 = REPLAY_START + timedelta(days=n)
+        prices = synth_day(n)
+        plan = plan_of(prices, start=day0)
+        open_buy = None
+        for h in range(24):
+            d = act(plan, h, thresh=threshold, unc=unc, soc=soc, floor=floor,
+                    ceiling=ceiling, start=day0)
+            if d.action is None:
+                continue
+            price = prices[h]
+            if d.action == CHARGE:
+                soc = min(ceiling, soc + STEP_PCT)
+                open_buy = (price, d)
+            else:
+                if price < 0:
+                    neg_sells += 1
+                soc = max(floor, soc - STEP_PCT)
+                if open_buy is not None:
+                    buy_price, buy_d = open_buy
+                    trades.append((buy_price, price, d))
+                    open_buy = None
+                else:
+                    holdings.append((n, h, price))
+            if d.spread_eur_kwh is None or d.required_eur_kwh is None:
+                bad.append((n, h))
+    return trades, holdings, bad, neg_sells
+
+
+_trades, _held_over, _bad, _neg_sells = derived(
+    lambda: replay(0.05, NO_WEATHER)) or (None, None, None, None)
+_kwh = MAX_W / 1000.0
+check(f"the replay is 365 days long and actually traded ({len(_trades)} "
+      f"completed round trips), or every invariant below would hold vacuously",
+      len(_trades) > 30)
+check("...and every trade carries the spread and the threshold it cleared, so "
+      "the ledger is auditable rather than a count",
+      _bad == [])
+check("...and NOT every day traded, so the strategy is selective rather than "
+      "unconditional - a replay in which everything trades grades nothing",
+      len(_trades) < 365)
+_losses = [(b, s) for b, s, _ in _trades if s - b < 0.05]
+check(f"EVERY completed round trip cleared the threshold it was run at: sell "
+      f"minus buy >= 0.05 EUR/kWh on all {len(_trades)} of them "
+      f"({len(_losses)} exceptions)", _losses == [])
+_under = [d.spread_eur_kwh for _b, _s, d in _trades
+          if d.spread_eur_kwh < d.required_eur_kwh]
+check("...and each one cleared its own WIDENED requirement, which is the "
+      "stricter statement and the one §8.3 actually asks for",
+      _under == [])
+_net = sum((s - b - 0.05) * _kwh for b, s, _ in _trades)
+check(f"so the replayed year is net-positive BY CONSTRUCTION, not by luck: "
+      f"{_net:+.2f} EUR of benefit net of the cycle cost over "
+      f"{len(_trades)} trades at {_kwh:.1f} kWh each, and no single trade "
+      f"contributes a negative term", _net > 0)
+check(f"§17 held for the whole year: {_neg_sells} discharges into a "
+      f"negative-price hour", _neg_sells == 0)
+# ...and the replay really did contain negative hours, or the line above is
+# a count of something that never had the chance to happen.
+_neg_hours = sum(1 for n in range(365) for p in synth_day(n) if p < 0)
+check(f"...and it had the chance to fail: {_neg_hours} hours of the synthetic "
+      f"year are priced below zero", _neg_hours > 100)
+# The shipped default, over the same year. Reported, not tuned.
+_sh_trades = derived(lambda: replay(_shipped, NO_WEATHER)[0])
+print(f"     [report] at the SHIPPED threshold {_shipped} EUR/kWh widened by "
+      f"{widen_factor(NO_WEATHER)}x, the same synthetic year completes "
+      f"{len(_sh_trades)} round trips; at 0.05 it completes {len(_trades)}. "
+      f"The shipped default is deliberately conservative and the number that "
+      f"matters is the one a REAL year would give, which this repository "
+      f"cannot answer.")
+check("the shipped threshold is strictly more selective than the loose one on "
+      "the same year - the parameter does what it says, at whatever value the "
+      "owner sets it to",
+      len(_sh_trades) < len(_trades))
+check("AC4 IS REPORTED AS NOT MET rather than redefined: this file says so in "
+      "section 7's preamble, and there is no price history in the repository "
+      "to meet it with",
+      not any(f.endswith((".csv", ".json.gz"))
+              and "price" in f.lower()
+              for f in os.listdir(os.path.join(HERE, "scenarios"))))
+
+
+# --------------------------------------------------------------------------- #
+print()
+print("8. the background refresh - the tick must never await a fetch")
+
+
+_MAIN_SRC = derived(lambda: open(os.path.join(HERE, "app", "main.py"),
+                                 encoding="utf-8").read())
+_cycle = derived(lambda: _fn(_MAIN_SRC, "cycle"))
+_refresh = derived(lambda: _fn(_MAIN_SRC, "run_tariff_refresh"))
+_read_inputs = derived(lambda: _fn(_MAIN_SRC, "read_inputs"))
+_FETCHERS = {"resolve_tariff", "load_prices", "https_get"}
+check("app/main.py has a run_tariff_refresh() coroutine at all",
+      _refresh is not None)
+check("...and IT is what awaits resolve_tariff(), so the fetch has somewhere "
+      "to live that is not the control path",
+      isinstance(_refresh, ast.AsyncFunctionDef)
+      and "resolve_tariff" in _awaited(_refresh))
+# ⚠️ THE FORWARD HAZARD, AND IT IS THE ONE THE CARD CALLS THE MOST DANGEROUS
+# THING IN THIS TICKET. cycle() and read_inputs() are both on the path that
+# feeds the ESP32 heartbeat; layer 1 commands 0 W if that stops for ~30 s. A
+# supplier's TLS timeout on either of them is a SUPPLIER able to drive this
+# add-on into the failsafe. Nothing warns about it at runtime - the add-on would
+# simply be slow - so it is asserted structurally, on the shipped AST.
+check(f"cycle() awaits NONE of {sorted(_FETCHERS)} - a supplier's TLS timeout "
+      f"on the tick is a supplier able to push this add-on towards layer 1's "
+      f"30 s veto {sorted(_awaited(_cycle))}",
+      isinstance(_cycle, ast.AsyncFunctionDef)
+      and not (_awaited(_cycle) & _FETCHERS))
+check("...and neither does read_inputs(), which is on the same path and is "
+      "where a fetch would most naturally be put",
+      isinstance(_read_inputs, ast.AsyncFunctionDef)
+      and not (_awaited(_read_inputs) & _FETCHERS))
+check("...and the check above can fail: the scanner really does find the "
+      "awaits that ARE there",
+      "resolve_tariff" in _awaited(_refresh)
+      and len(_awaited(_cycle)) > 0)
+
+# The age bound, driven on the shipped expression rather than a copy of it.
+class _Stub:
+    """Just enough Controller for fresh_tariff_plan(), which touches nothing else."""
+    fresh_tariff_plan = _main.Controller.fresh_tariff_plan
+
+    def __init__(self, plan, at):
+        self.tariff_plan, self.tariff_at = plan, at
+
+
+_NOW = datetime(2026, 6, 1, 12, 0, tzinfo=UTC)
+_fresh_s = _main.TARIFF_MAX_AGE_S
+# ⚠️ THE ABSOLUTE PROBE COMES FIRST, AND THE FIRST DRAFT SHOWED WHY. Probing at
+# `TARIFF_MAX_AGE_S + 1` reads the constant under test, so widening the constant
+# moves the probe with it and the check passes for a bound that has been made
+# unreachable - a mutation multiplying it by a million SURVIVED that draft.
+# Twelve hours is stale by any reading of a feed that publishes daily, and it is
+# stated here rather than derived from the module being graded.
+check("a plan resolved TWELVE HOURS ago is not handed to the strategy, whatever "
+      "the module's own constant says - a stale price is not a fresh one",
+      _Stub(PLAN, _NOW - timedelta(hours=12)).fresh_tariff_plan(_NOW) is None)
+check(f"...and the constant is inside the range that makes that true: "
+      f"{_fresh_s:.0f} s is under six hours and over one refresh interval",
+      _main.TARIFF_REFRESH_S < _fresh_s < 6 * 3600)
+check("a plan resolved a moment ago IS handed to the strategy, so the two "
+      "lines above are a bound rather than a blanket refusal",
+      _Stub(PLAN, _NOW - timedelta(seconds=1)).fresh_tariff_plan(_NOW) is PLAN)
+check(f"...and the bound is where the module says it is: one second past "
+      f"{_fresh_s:.0f} s fails and one second short of it passes",
+      _Stub(PLAN, _NOW - timedelta(seconds=_fresh_s + 1)
+            ).fresh_tariff_plan(_NOW) is None
+      and _Stub(PLAN, _NOW - timedelta(seconds=_fresh_s - 1)
+                ).fresh_tariff_plan(_NOW) is PLAN)
+check("a plan with no timestamp is not fresh either - both terms are required, "
+      "and 'nothing stored it' is not 'it is current'",
+      _Stub(PLAN, None).fresh_tariff_plan(_NOW) is None
+      and _Stub(None, _NOW).fresh_tariff_plan(_NOW) is None)
+check("the refresh cadence is strictly under the age bound, or the plan would "
+      "expire between two healthy refreshes and arbitrage would flicker",
+      _main.TARIFF_REFRESH_S < _main.TARIFF_MAX_AGE_S)
+# CFG-03: main.py gained no fourth copy of an option default.
+_names = derived(lambda: {n.id for n in ast.walk(ast.parse(_MAIN_SRC))
+                          if isinstance(n, ast.Name)})
+check("main.py reads every arbitrage option through app/arbitrage.py or "
+      "price_option(), never through `o.get('cycle_cost_threshold_eur', <a "
+      "number>)` - CFG-03 still tracks three places, not four",
+      all(f'"{k}"' not in _MAIN_SRC and f"'{k}'" not in _MAIN_SRC
+          for k in ("cycle_cost_threshold_eur", "arbitrage_soc_floor",
+                    "arbitrage_soc_ceiling")))
+
+
+# --------------------------------------------------------------------------- #
+print()
+print("9. the startup gate - the configurations that cannot work are refused")
+
+
+def vbase(**over):
+    o = {"setpoint_entity": "number.goodwe_setpoint", "max_w": 2000,
+         "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
+         "meter_source": "off",
+         "maintenance_charge_w": 2000, "maintenance_discharge_w": 2000}
+    o.update(over)
+    return o
+
+
+def why(problems, needle):
+    return any(needle in p for p in problems)
+
+
+P1_ON = {"meter_source": "ha_signed", "p1_net_entity": "sensor.p1"}
+check("the shipped configuration still starts - arbitrage_enabled ships off "
+      "and this ticket may not have made an existing install unbootable",
+      validate_startup_config(vbase(), -5000, 5000) == []
+      and enabled({}) is False)
+check("arbitrage on with a provider and P1 starts",
+      validate_startup_config(
+          vbase(arbitrage_enabled=True, price_provider="generic_dayahead",
+                **P1_ON), -5000, 5000) == [])
+check("...and so does arbitrage on a static profile with P1, because §9.2 "
+      "makes a static site first-class rather than unsupported",
+      validate_startup_config(
+          vbase(arbitrage_enabled=True, static_profile=STATIC_BE, **P1_ON),
+          -5000, 5000) == [])
+check("arbitrage on with NO price source at all is refused - a strategy that "
+      "must refuse on every cycle forever looks exactly like one that is "
+      "running and finding nothing",
+      why(validate_startup_config(vbase(arbitrage_enabled=True, **P1_ON),
+                                  -5000, 5000), "NO PRICES AT ALL"))
+check("arbitrage on with meter_source off is refused - §9.1 puts peak shaving "
+      "above arbitrage and an unmeasured peak cannot be shaved",
+      why(validate_startup_config(
+          vbase(arbitrage_enabled=True, price_provider="generic_dayahead"),
+          -5000, 5000), "meter_source"))
+check("a floor at or above the ceiling is refused - there is no band left "
+      "between them and the strategy could never act",
+      why(validate_startup_config(
+          vbase(arbitrage_enabled=True, price_provider="generic_dayahead",
+                arbitrage_soc_floor=90, arbitrage_soc_ceiling=90, **P1_ON),
+          -5000, 5000), "no band left"))
+check("...and the same numbers the right way round start",
+      validate_startup_config(
+          vbase(arbitrage_enabled=True, price_provider="generic_dayahead",
+                arbitrage_soc_floor=25, arbitrage_soc_ceiling=90, **P1_ON),
+          -5000, 5000) == [])
+check("a NEGATIVE cycle_cost_threshold_eur is refused even with arbitrage OFF "
+      "- it is a cost, and a negative one is a threshold every spread clears",
+      why(validate_startup_config(vbase(cycle_cost_threshold_eur=-0.01),
+                                  -5000, 5000), "non-negative")
+      and raises(lambda: threshold_eur_kwh({"cycle_cost_threshold_eur": -1}),
+                 ValueError))
+# Rule 7's wording, which OPT-04 had to update because it named what main.py
+# did NOT do. test_price.py section 12 pins the load_prices consumer set; this
+# pins the sentence that set exists to keep honest.
+_V = derived(lambda: open(os.path.join(HERE, "app", "validate.py"),
+                          encoding="utf-8").read())
+check("rule 7's message no longer says a configured provider shapes no energy "
+      "decision - OPT-04 made that false, and the instruction attached to the "
+      "tripwire was to fix the wording rather than to silence the check",
+      isinstance(_V, str)
+      and "shape no energy decision" not in _V
+      and "still shapes no energy decision" not in _V)
+check("...and it names the consumer that now exists, so it cannot decay into a "
+      "refusal with no stated reason",
+      isinstance(_V, str) and "app/arbitrage.py" in _V)
+# The option mirror, pinned to config.yaml the way TARIFF-02 pinned the rest.
+_CFG = derived(lambda: open(os.path.join(HERE, "config.yaml"),
+                            encoding="utf-8").read())
+check("every option this card added is in config.yaml's options block AND its "
+      "schema block, or the Supervisor drops it and the add-on silently runs "
+      "on the mirror instead of the operator's value",
+      isinstance(_CFG, str)
+      and all(_CFG.count(f"\n  {k}:") == 2
+              for k in ("arbitrage_enabled", "cycle_cost_threshold_eur",
+                        "arbitrage_soc_floor", "arbitrage_soc_ceiling")))
+check("...and app/arbitrage.py reads them through price_option() only - one "
+      "read path, one place the defaults live",
+      "PRICE_OPTION_DEFAULTS[" not in
+      open(os.path.join(HERE, "app", "arbitrage.py"), encoding="utf-8").read())
+check("soc_bounds() returns the shipped pair, floor first",
+      soc_bounds({}) == (float(PRICE_OPTION_DEFAULTS["arbitrage_soc_floor"]),
+                         float(PRICE_OPTION_DEFAULTS["arbitrage_soc_ceiling"])))
+
+
+# --------------------------------------------------------------------------- #
+print()
+if fails:
+    print(f"{len(fails)} FAILED: " + "; ".join(fails))
+    sys.exit(1)
+print("all checks passed")

goodwe_controller/test_price.py

@@ -212,6 +212,14 @@ def shipped(key):
         if in_schema or not line.startswith(f"  {key}:"):
             continue
         raw = line.split(":", 1)[1].split("#", 1)[0].strip().strip('"').strip("'")
+        # OPT-04 added the first BOOLEAN price option, and without this the
+        # mirror check would compare the string "false" against `False` and go
+        # red for a file that is perfectly in step. Parsing a bool as a bool is
+        # what the yaml module this image does not carry would have done; it
+        # widens nothing, because a key whose shipped value disagrees with the
+        # mirror still drifts.
+        if raw in ("true", "false"):
+            return raw == "true"
         try:
             return int(raw)
         except ValueError:

goodwe_controller/test_price_forecast.py

@@ -71,6 +71,7 @@ from app.price_forecast import (
     MAINTENANCE_CHARGE, MAINTENANCE_DRAIN, Gate, Irreversible, committed_hours,
     fill_from_yesterday, price_permits, begin_phase,
 )
+from app.arbitrage import ArbitrageDecision
 from app.static_tariff import Arbitrage, TariffPlan
 
 UTC = timezone.utc
@@ -582,7 +583,13 @@ _ctors = {"Gate.permit": Gate.permit, "Gate.block": Gate.block,
           "Arbitrage.allow": Arbitrage.allow, "Arbitrage.refuse": Arbitrage.refuse,
           "Arbitrage.match": Arbitrage.match,
           "_Allowed.__init__": _st._Allowed.__init__,
-          "_Refused.__init__": _st._Refused.__init__}
+          "_Refused.__init__": _st._Refused.__init__,
+          # ⚠️ OPT-04's outcome type is scanned HERE, in the same dict, rather
+          # than by a hand-written case in test_arbitrage.py - which is TEST-14's
+          # actual instruction. A new outcome type extends the SCAN; a
+          # hand-written case is the thing that gets forgotten, and the whole
+          # reason this dict exists instead of five separate checks.
+          "ArbitrageDecision.__init__": ArbitrageDecision.__init__}
 _with_defaults = derived(lambda: {n: d for n, d in
                                   ((n, _defaulted(f)) for n, f in _ctors.items())
                                   if d})
@@ -835,13 +842,28 @@ print("7. CFG-03 and validate.py rule 7 - what this ticket did NOT widen")
 _defaults = derived(lambda: sorted(_price.PRICE_OPTION_DEFAULTS))
 _cfg = derived(lambda: open(os.path.join(HERE, "config.yaml"),
                             encoding="utf-8").read())
-check(f"FCST-03 added ZERO option keys: the in-code mirror still holds exactly "
-      f"TARIFF-02's and TARIFF-03's {_defaults}",
+# ⚠️ THIS LIST IS A DRIFT PIN, NOT A BAN ON EVER ADDING AN OPTION. FCST-03 added
+# none, and said so here. OPT-04 added FOUR - and added them to this ONE dict
+# and to config.yaml, which is precisely the behaviour CFG-03 asks for: the
+# alternative is a fourth freehand copy in main.py's `o.get(k, X)` fallbacks.
+# The pin is what makes such an addition a decision somebody signs rather than a
+# thing that appears; whoever adds the fifth comes here and says which card and
+# why, as this line does.
+_OPT04_KEYS = ["arbitrage_enabled", "cycle_cost_threshold_eur",
+               "arbitrage_soc_floor", "arbitrage_soc_ceiling"]
+check(f"the in-code mirror holds exactly TARIFF-02's, TARIFF-03's and OPT-04's "
+      f"keys - FCST-03 still added none of its own {_defaults}",
       _defaults == sorted(["price_provider", "price_markup_eur_kwh",
                            "price_taxes_eur_kwh", "export_price_eur_kwh",
                            "price_csv_path", "price_feed_url",
                            "price_horizon_h", "price_timeout_s",
-                           "static_profile"]))
+                           "static_profile"] + _OPT04_KEYS))
+# ...and the four are genuinely NEW, not a rename of something that was already
+# there, or the line above would pass while the mirror silently lost a key.
+check("...and OPT-04's four keys are additions rather than replacements: the "
+      "nine TARIFF-02/03 keys are all still present",
+      isinstance(_defaults, list)
+      and len(set(_defaults) - set(_OPT04_KEYS)) == 9)
 check("...and every one of them is still in config.yaml, which stays the "
       "source of truth",
       isinstance(_cfg, str)