"""Runnable check for OPT-15, the P3 price-run fixes. `python3 test_price_run_fixes.py` The OPT-11 P3 post-merge test on TEST-23 (5a4cdc2) found three defects. This file drives the pure modules and the REAL Controller (tools/opt11p3_replay.py) with the lagged model (`Lag`: command dead time, meter and battery reading age, the P1 debounce, 1 s cycles), which reproduces all three on 5a4cdc2. 1. F1: a quarter edge while charging does not stop the charge (the guard projects the quarter `now` is in), and a real guard stop still latches. 2. F2: the export overshoot at a charge end stays within 0.15 x rating in the model and in the lagged variants (TEST-23 measured -1120 W). 3. F3: a charge refusal inside a cheap run is probed every 15 minutes. 4. The P3 and SAFETY-79 invariants on the lagged runs. 5. Modes off is byte-identical to 5a4cdc2, also with a late P1 sample on every quarter boundary, and the harness sees a change there. DECISIONS/OPT-11-battery-modes.md has the record. Mutations (MUTATION.md, purged copies, graded over this file and test_modes_p3.py) are listed in the pull request. """ import logging import os import sys from dataclasses import replace from datetime import datetime, timedelta, timezone from types import SimpleNamespace HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) sys.path.insert(0, os.path.join(HERE, "tools")) logging.disable(logging.WARNING) import opt11p3_identity as ident # noqa: E402 import opt11p3_replay as rp # noqa: E402 from app import arbitrage, modetrade, packmode, peakguard # noqa: E402 from app import tariff # noqa: E402 from app import control as ctl # noqa: E402 from app import main # noqa: E402 from app.p1 import QuarterAverager, make_sample # noqa: E402 from app.tariff import PeakTracker # noqa: E402 from named_absence import nth, present # noqa: E402 fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) L = timezone(timedelta(hours=2)) A, B = "inverter-a", "inverter-b" D0 = datetime(2026, 9, 9, 0, 0, tzinfo=L) UTC0 = D0.astimezone(timezone.utc) # Cheap 02:00-04:00 local, dear 19:00-20:00, flat otherwise (two days). PRICES = [0.30] * 192 for _i in range(8, 16): PRICES[_i] = 0.10 for _i in range(76, 80): PRICES[_i] = 0.40 WPTS = [(UTC0 + timedelta(minutes=15 * i), 900, p) for i, p in enumerate(PRICES)] RATING = rp.RATING BOUND = 0.15 * RATING AIM = rp.PEAK_CAP - 50 CH_SF = {"a": (True, packmode.CHEAP_HOURS, False), "b": (True, packmode.SOLAR_FIRST, False)} def at(h): return UTC0 + timedelta(hours=h) def house(steps, base=800.0): """[(t, W)] for the replay: `base`, then each (local hour, W) from then.""" out = [(UTC0.timestamp() - 10, base)] out += [(at(h).timestamp(), float(w)) for h, w in steps] return out def run(start_h, hours, *, lag=None, soc_a=20.0, steps=(), choice=None, refuse=(), patch=None, check_cb=None): rows = [] def cb(c, h, now, pr): sh = dict(c.allocation.shares) if c.allocation is not None and \ c.allocation.shares else {} rows.append(dict( now=now, grid=h.grid(), shares=sh, action=getattr(c.arb, "action", None), reason=getattr(c.arb, "reason", None) or "", roles=dict(c.mode_tick.roles) if c.mode_tick else {}, refused=c.inverters[0].refused, probing=c.inverters[0].probe_from is not None, headroom=c.arb_headroom_w, p={u: p.p for u, p in h.plants.items()})) if check_cb: check_cb(c, h, now, pr) st, c = rp.run("p3_ch_sell", house(steps), WPTS, at(start_h), 3600 * hours, check=cb, lag=lag, soc0={"a": soc_a, "b": 60.0}, choice=CH_SF if choice is None else choice, refuse=refuse, patch=patch, extra={"cycle_cost_threshold_eur": 0.10}) return rows, c def stops(rows): """[(row index, row)] where a grid charge stopped.""" return [(i, rows[i]) for i in range(1, len(rows)) if rows[i - 1]["action"] == arbitrage.CHARGE and rows[i]["action"] != arbitrage.CHARGE] def q(when): return int(when.timestamp() // 900) def banked(c, rows): """The add-on's own P1 blocks for the quarters that had a grid charge.""" chq = {q(r["now"]) for r in rows if r["action"] == arbitrage.CHARGE} return [b.offtake_avg_w for b in c.p1.blocks if q(b.start) in chq] def mixed(rows): return [r for r in rows if any(v > 0 for v in r["shares"].values()) and any(v < 0 for v in r["shares"].values())] # Close to TEST-23's plant: 1 s cycles, readings 1 s old, a battery that # follows the previous second's command. LAG = rp.Lag(tau_s=0.05, ramp_w_s=1e9) # ============================================================================ # print("1. F1: a quarter edge while charging") # 1a. The projection on the edge tick, before the new block's first sample. _avg = QuarterAverager(phases=1, max_hold_s=30.0) _edge = datetime(2026, 9, 9, 12, 0, tzinfo=timezone.utc) for _s in range(-900, 1, 1): _t = _edge + timedelta(seconds=_s - 0.35) _avg.add(make_sample("ha_signed", 3300.0, 0.0, phases=1, ingest_ts=_t, ingest_mono=0.0)) _now = _edge + timedelta(seconds=0.6) _old = PeakTracker.project(_avg, _now) _new = PeakTracker.project(_avg, _now, roll=True) def _decide(pr): return peakguard.decide(pr, batt_w=-2500.0, cap_w=3500.0, margin_w=50.0, reading=None, achievable_w=5000.0, soc=50.0, soc_floor=20.0, max_hold_s=30.0) print(f" edge tick: old block {_old.explain()}; rolled {_new.explain()}") check("on the edge tick the open block is still the old one: the guard reads " "'no seconds left' and the headroom is unmeasured (the F1 cause)", _old.elapsed_s >= 900 and peakguard.charge_headroom_w( _decide(_old)) is None) _later = QuarterAverager(phases=1, max_hold_s=30.0) for _s in range(-900, 1, 1): _later.add(make_sample("ha_signed", 3300.0, 0.0, phases=1, ingest_ts=_edge + timedelta(seconds=_s - 0.35), ingest_mono=0.0)) _later.add(make_sample("ha_signed", 3300.0, 0.0, phases=1, ingest_ts=_edge + timedelta(seconds=0.6), ingest_mono=0.0)) _real = PeakTracker.project(_later, _now) _hr = peakguard.charge_headroom_w(_decide(_new)) check("rolled, it is the new quarter, measured, and equal to what the " "averager projects once the new block's first sample is in; the " "headroom is the house and the aim (3450 - 800 W)", _new.block_start == _edge == _real.block_start and _new.complete and abs(_new.projected_w - _real.projected_w) < 0.01 and abs(_new.elapsed_s - 0.6) < 1e-6 and _hr is not None and abs(_hr - 2650.0) < 1.0) _gap = QuarterAverager(phases=1, max_hold_s=30.0) _gap.add(make_sample("ha_signed", 3300.0, 0.0, phases=1, ingest_ts=_edge - timedelta(seconds=100), ingest_mono=0)) _gp = PeakTracker.project(_gap, _edge + timedelta(seconds=40), roll=True) check("...and a sample older than the hold is not credited: past the hold " "the rolled quarter is blind, as the averager would make it", _gp.elapsed_s == 40.0 and _gp.covered_frac == 0.0 and peakguard.blindness(_gp, 30.0, False) is not None) check("without `roll` nothing changes (the modes-off path)", PeakTracker.project(_avg, _now, roll=False) == _old) # 1b. The lagged replay: 02:00-03:30 local, a Cheap hours battery charging # under an 800 W house through five quarter edges. ROWS1, C1 = run(1.9, 1.6, lag=LAG, choice={ "a": (True, packmode.CHEAP_HOURS, False), "b": (True, packmode.LEFTOVER_SOLAR, False)}) _ch1 = [r for r in ROWS1 if r["action"] == arbitrage.CHARGE] _edges1 = sorted({q(r["now"]) for r in _ch1}) _st1 = stops(ROWS1) print(f" {len(_ch1)} charge cycles over {len(_edges1)} quarters; stops " f"{[(r['now'].astimezone(L).strftime('%H:%M:%S'), r['reason'][:40]) for _, r in _st1]}") check("no stop at any quarter edge crossed while charging (5a4cdc2 in this " "model: a stop at each, 'no trade (headroom)', latched for the quarter)", len(_edges1) >= 5 and not _st1) def _no_roll(c): real = c.peak.project c.peak.project = lambda a, now, roll=False: real(a, now) ROWS1M, _ = run(1.9, 0.6, lag=LAG, patch=_no_roll) _m1 = stops(ROWS1M) check("non-vacuous: with the old projection this model stops at the edge " "for want of a measured headroom (5a4cdc2 then latched the quarter; " "the latch fix alone keeps it to that tick)", _m1 and "headroom" in _m1[0][1]["reason"] and _m1[0][1]["now"].astimezone(L).strftime("%M:%S") == "15:00") # 1c. A real guard stop still latches, and the 500 W minimum stays. # House: 800 W; 3700 W at 02:20-02:21; 3000 W from 02:26 (headroom 450 W # in the next quarter); 2850 W from 02:40 (600 W). ROWS2, C2 = run(1.9, 0.9, lag=LAG, steps=[ (2 + 20 / 60, 3700), (2 + 21 / 60, 800), (2 + 26 / 60, 3000), (2 + 40 / 60, 2850)]) _st2 = stops(ROWS2) _first = _st2[0][1] if _st2 else None _q2 = q(_first["now"]) if _first else None _after2 = [r for r in ROWS2 if _first and r["now"] > _first["now"] and q(r["now"]) == _q2] print(f" guard stop at " f"{_first['now'].astimezone(L).strftime('%H:%M:%S') if _first else None}" f" ({_first['reason'][:40] if _first else ''})") check("a mid-quarter 3.7 kW load stops the charge for the peak and it stays " "off for the rest of that quarter, the house back at 800 W", _first is not None and _first["now"].astimezone(L).minute == 20 and ("peak" in _first["reason"] or "headroom" in _first["reason"]) and _after2 and all(r["action"] is None for r in _after2)) _starts = [b for a, b in zip(ROWS1 + ROWS2, (ROWS1 + ROWS2)[1:]) if a["action"] != arbitrage.CHARGE and b["action"] == arbitrage.CHARGE] _low = [r for r in ROWS2 if r["headroom"] is not None and 0 < r["headroom"] <= modetrade.START_W and r["action"] is None] print(f" {len(_starts)} charge starts, lowest headroom at a start " f"{min(r['headroom'] for r in _starts):.0f} W; {len(_low)} cycles " f"with 0-500 W of headroom and no charge") check("the 500 W minimum stays: every charge starts with more than 500 W of " "headroom, and a house at 3 kW (450 W) holds it off", _starts and _low and all(r["headroom"] > modetrade.START_W for r in _starts)) # 1d. An unmeasured headroom stops one tick and does not latch. _c1d = run(1.9, 0.2, soc_a=20.0)[1] _n1d = UTC0 + timedelta(hours=2, minutes=5) _ok1d = False if _c1d.mode_tick is not None and _c1d.mode_tick.action == modetrade.CHARGE: _c1d.arb_headroom_w = None _t1 = _c1d._mode_roles(_n1d, None) _c1d.arb_headroom_w = 3000.0 _t2 = _c1d._mode_roles(_n1d + timedelta(seconds=1), None) _c1d.arb_headroom_w = 0.0 _t3 = _c1d._mode_roles(_n1d + timedelta(seconds=2), None) _c1d.arb_headroom_w = 3000.0 _t4 = _c1d._mode_roles(_n1d + timedelta(seconds=3), None) _ok1d = (_t1.action is None and _t2.action == modetrade.CHARGE and _t3.action is None and _t4.action is None and _c1d._charge_blocked_q == q(_n1d)) check("an unmeasured headroom (None) stops the charge that tick only; a " "measured 0 W stops it for the rest of the quarter", _ok1d) # ============================================================================ # print("\n2. F2: the export overshoot at a charge end") # A fills to its 95 % ceiling at about 04:03 under an 800 W house, and # Solar first B takes the house over. VARIANTS = { "model (5 s, instant readings)": None, "model, lagged (1 s cycles)": rp.Lag(), "TEST-23 plant, meter 1 s": rp.Lag(tau_s=0.05, ramp_w_s=1e9), "TEST-23 plant, meter 2 s": rp.Lag(tau_s=0.05, ramp_w_s=1e9, meter_s=2.0), "meter 2 s, battery reading fresh": rp.Lag(tau_s=0.05, ramp_w_s=1e9, meter_s=2.0, batt_s=0.0), "2 s cycles, meter 2 s": rp.Lag(tick=2.0, meter_s=2.0), "5 s meter": rp.Lag(tick=5.0, meter_s=5.0, batt_s=2.0), "slow battery (8 s, 2 s dead), 5 s meter": rp.Lag( tau_s=8.0, dead_s=2.0, tick=5.0, meter_s=5.0, batt_s=2.0), } def charge_end(rows, lag): """(min grid, max grid, pre-edge grid, settle s) after the first charge end, over 120 s; settle: from then on within 40 W of the target.""" st = stops(rows) if not st: return None i = st[0][0] tick = rp.TICK if lag is None else lag.tick w = [r["grid"] for r in rows[i:i + int(120 / tick)]] settle = next((j * tick for j in range(len(w)) if all(abs(x + 10) < 40 for x in w[j:])), None) return min(w), max(w), rows[i - 1]["grid"], settle F2 = {} for _k, _lag in VARIANTS.items(): F2[_k] = charge_end(run(3.9, 0.5, lag=_lag, soc_a=86.0)[0], _lag) print(f" {_k}: {F2[_k] and tuple(round(x) if x is not None else x for x in F2[_k])}") check(f"every variant: the charge ends, the export overshoot stays within " f"0.15 x rating ({BOUND:.0f} W), no import above the charge's own, and " f"the grid settles within 45 s (90 s for the slow battery)", all(v is not None and v[0] >= -BOUND and v[1] <= v[2] + 50 and v[3] is not None and v[3] <= (90 if k.startswith("slow") else 45) for k, v in F2.items())) def _no_hold(c): c._edge_grid = lambda end, now: c.grid _pre = charge_end(run(3.9, 0.5, lag=VARIANTS["TEST-23 plant, meter 2 s"], soc_a=86.0, patch=_no_hold)[0], VARIANTS["TEST-23 plant, meter 2 s"]) print(f" without the hold (5a4cdc2's reseed), meter 2 s: " f"{_pre and tuple(round(x) if x is not None else x for x in _pre)}") check("non-vacuous: without the hold the lagged model shows TEST-23's " "overshoot (over 0.4 x rating), so it cannot hide again", _pre is not None and _pre[0] < -0.4 * RATING) # The hold must not hide a real load change at the edge. _down = run(3.9, 0.5, lag=LAG, soc_a=86.0, steps=[])[0] _i_end = stops(_down)[0][0] if stops(_down) else None _edge_t = _down[_i_end]["now"] if _i_end else None def _edge_step(w): """The same charge end with the house stepping to `w` at the edge: (seconds until the loop moves B off E, lowest grid, highest grid after the first 3 s).""" if _edge_t is None: return None h = (_edge_t - UTC0).total_seconds() / 3600 rows = run(3.9, 0.5, lag=LAG, soc_a=86.0, steps=[(h, w)])[0] after = [r for r in rows if r["now"] >= _edge_t] first = nth(after, 0, "the edge cycle") if not present(first): return None e = first["shares"].get(B) moved = next(((r["now"] - _edge_t).total_seconds() for r in after if r["shares"].get(B) != e), None) return moved, min(r["grid"] for r in after[:60]), max( r["grid"] for r in after[3:60]), [r["grid"] for r in after[:60]] _fall, _rise = _edge_step(200.0), _edge_step(1800.0) print(f" house 800 -> 200 W at the edge: loop moves after {_fall[0]} s, " f"lowest {_fall[1]:.0f} W; 800 -> 1800 W: after {_rise[0]} s, highest " f"{_rise[2]:.0f} W") check("the hold does not hide a load change: a house that falls at the edge " "ends it within 5 s, with no export beyond the fall and the bound; one " "that rises is followed by EDGE_HOLD_S, with no import beyond the rise", _fall and _rise and _fall[0] is not None and _fall[0] <= 5 and _fall[1] >= -600 - BOUND and _rise[0] is not None and _rise[0] <= main.EDGE_HOLD_S + 2 and _rise[2] <= 1000 + 50) _peak = _edge_step(4500.0) _g10 = nth(_peak[3], 10, "10 s after the edge") if _peak else None print(f" 800 -> 4500 W at the edge: grid 10 s later {_g10} W") check("...and when the peak floor takes the command during the hold, the " "hold ends: 10 s after the edge B gives its rating (grid 4500 - 2500 W), " "not the aim the floor holds", present(_g10) and _g10 <= 2100) # ============================================================================ # print("\n3. F3: a charge refusal inside a cheap run is probed") # A refuses to charge from its first charge; the plant accepts again at # 02:40. The cheap run lasts to 04:45. def _accept(c, h, now, pr): if now >= at(2 + 40 / 60): h.plants["a"].refuse_charge = False ROWS3, C3 = run(1.9, 1.2, soc_a=30.0, refuse=("a",), check_cb=_accept) _ref3 = [r for r in ROWS3 if r["refused"]] _pr3 = [r for r in ROWS3 if r["probing"] and r["action"] == arbitrage.CHARGE] _probe_q = sorted({r["now"].astimezone(L).strftime("%H:%M") for r in _pr3}) _end3 = next((r for r in ROWS3 if _ref3 and r["now"] > _ref3[-1]["now"]), None) print(f" refused {len(_ref3)} cycles; charge probes at {_probe_q}; " f"after: {_end3 and _end3['now'].astimezone(L).strftime('%H:%M:%S')}" f" {_end3 and _end3['action']}") check("while A refuses, it is probed every 15 minutes: at PROBE_W exactly, " "with B resting (no share on B, one sign)", len(_probe_q) >= 2 and all( r["shares"].get(A) == -ctl.PROBE_W and r["shares"].get(B, 0) == 0 for r in _pr3)) check("...outside a probe the refusal wins: no grid trade and A never " "charges", all(r["action"] is None and r["shares"].get(A, 0) >= 0 for r in _ref3 if not r["probing"])) check("...and the probe after the plant accepts again ends the refusal and " "the charge resumes in the same run", _end3 is not None and _end3["now"] < at(3) and any(r["action"] == arbitrage.CHARGE and r["shares"].get(A, 0) < -1000 for r in ROWS3 if r["now"] > _end3["now"])) _pk = modetrade.Pack(A, True, packmode.CHEAP_HOURS, False, 2500.0, 30.0, 10.0, refused=-1, probing=True) _runs = SimpleNamespace(on=True, cheap=SimpleNamespace(holds=lambda n: True), charge_useful=lambda: True, give_at=lambda n: None, sell=None) def _dec(packs, **kw): args = dict(charge_ok=True, sell_ok=True, trade_why=None, price_now=0.1, headroom_w=3000.0, floor_pct=20.0, ceiling_pct=95.0, house_w=800.0) args.update(kw) return modetrade.decide(_runs, UTC0, packs, **args) _pb = modetrade.Pack(B, True, packmode.SOLAR_FIRST, False, 2500.0, 60.0, 10.0) check("modetrade: a probing charge refuser is a charger, the others rest; " "not while in maintenance, not on Leftover solar or Solar first, not " "latched, not under the 500 W minimum, not with the guard engaged, " "and never in energy() (the plan's sizing)", _dec([_pk, _pb]).action == modetrade.CHARGE and _dec([_pk, _pb]).roles == {A: modetrade.CHARGE, B: modetrade.REST} and _dec([replace(_pk, idle=False), _pb]).action is None and all(_dec([replace(_pk, mode=m), _pb]).action is None for m in (packmode.LEFTOVER_SOLAR, packmode.SOLAR_FIRST)) and _dec([_pk, _pb], blocked=True).action is None and _dec([_pk, _pb], headroom_w=450.0).action is None and _dec([_pk, _pb], charge_ok=False, trade_why="peak").action is None and modetrade.energy([_pk], 20.0, 95.0)[1] == 0.0 and _dec([replace(_pk, probing=False), _pb]).action is None) # ============================================================================ # print("\n4. invariants on these runs (and part 1c's house in the model)") _rows4, _c4 = run(1.9, 0.9, steps=[ (2 + 20 / 60, 3700), (2 + 21 / 60, 800), (2 + 26 / 60, 3000), (2 + 40 / 60, 2850)]) _all = ROWS1 + ROWS2 + ROWS3 + _rows4 check("no opposite-direction split on any cycle", not mixed(_all)) _ceil = peakguard.quarter_ceiling_w(rp.PEAK_CAP, None)[0] _bk0 = banked(_c4, _rows4) _bk = banked(C1, ROWS1) + banked(C2, ROWS2) print(f" banked quarters with a grid charge: model {len(_bk0)}, highest " f"{max(_bk0):.2f} W; lagged {len(_bk)}, highest {max(_bk):.2f} W " f"(5a4cdc2: the same 3451.37 W); ceiling - margin {_ceil - 50:.0f} W") check("every banked 15-minute period with a grid charge is at most " "quarter_ceiling_w() - margin (SAFETY-79) in the model; lagged, the " "battery's lag may use 2 W of the 50 W margin, as on 5a4cdc2", _bk0 and max(_bk0) <= _ceil - 50 + 0.5 and _bk and max(_bk) <= _ceil - 50 + 2.0) check("Solar first and Leftover solar never charge from the grid: no charge " "role and no charge share on B while the site charges", all(r["roles"].get(B) != modetrade.CHARGE for r in _all) and all(r["shares"].get(B, 0) >= 0 for r in _all if r["action"] == arbitrage.CHARGE)) check("selling is off: no sale on any cycle", packmode.SELLING_ENABLED is False and all(r["action"] != arbitrage.DISCHARGE for r in _all)) # ============================================================================ # print("\n5. modes off: byte-identical to 5a4cdc2") # Printed by tools/opt11p3_identity.py on release/1.0 5a4cdc2 (2026-09-29), # goodwe-addon-checks:1, with this branch's harness (same API). AT_5A4CDC2 = { "n1": "71245b227ca40bada4f71f57ea6154475f4b0bd60239c9ad4f80bffb0094514c", "n2": "56cf28010b06765c29e5479abb815d7503d570b39240d82dfc818efa90c93722", "n1_lag": "c27ad6aada1dfed77351472b1944dae31639bbd5cb84f266e6741c97c3e59fc6", "n2_lag": "fafcae0e36718a640216e346be9f568c28126aae6e1ad2d18414f4ac74f90413", } _got = {k: ident.digest(ident.run(**ident.SCENARIOS[k])[0]) for k in AT_5A4CDC2} for _k in AT_5A4CDC2: print(f" {_k} {'same' if _got[_k] == AT_5A4CDC2[_k] else 'DIFFERS'} " f"{_got[_k][:16]}") check("with battery modes off every command, integrator, allocation and " "trade sentence equals 5a4cdc2's, also when every quarter boundary's " "cycle still sees the old P1 block", _got == AT_5A4CDC2) _real_project = tariff.PeakTracker.project tariff.PeakTracker.project = staticmethod( lambda a, now, roll=False: _real_project(a, now, roll=True)) try: _moved = ident.digest(ident.run(**ident.SCENARIOS["n2_lag"])[0]) \ != AT_5A4CDC2["n2_lag"] finally: tariff.PeakTracker.project = _real_project check("non-vacuous: the rolled projection leaking into modes off moves the " "late-sample scenario", _moved) print() if fails: print(f"{len(fails)} FAILED:") for f in fails: print(f" - {f}") sys.exit(1) print("all checks passed")