"""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. At two inverters SAFETY-91's `next_quarter()` already does that for the guard; at one inverter only the rolled projection does. 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 release/1.0 e3e4ba6, 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, n=2): 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, n=n, 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) # Review of #157 (X1): a held export is credited as 0 W, as `add()` banks it, # never as a negative that would pull the rolled quarter under the meter. _exp = QuarterAverager(phases=1, max_hold_s=30.0) for _s in range(-900, 1, 1): _exp.add(make_sample("ha_signed", 0.0, 1500.0, phases=1, ingest_ts=_edge + timedelta(seconds=_s - 0.35), ingest_mono=0.0)) _ep = PeakTracker.project(_exp, _now, roll=True) _exp.add(make_sample("ha_signed", 0.0, 1500.0, phases=1, ingest_ts=_edge + timedelta(seconds=0.6), ingest_mono=0.0)) _er = PeakTracker.project(_exp, _now) check("...and a held export rolls in as 0 W, equal to what the averager " "banks for those seconds", _ep.block_start == _er.block_start and _ep.projected_w == 0.0 and _ep.elapsed_w == 0.0 and _er.projected_w == 0.0) # 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 (two inverters; " "see below for the old projection)", 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) _CH1 = {"a": (True, packmode.CHEAP_HOURS, False), "b": (True, packmode.LEFTOVER_SOLAR, False)} # One inverter: SAFETY-91's `next_quarter()` is not applied there, so this is # where the old projection stops the charge at every edge. ROWS1N, C1N = run(1.9, 1.6, lag=LAG, n=1, choice=_CH1) ROWS1M, _ = run(1.9, 0.6, lag=LAG, n=1, patch=_no_roll, choice=_CH1) _m1 = stops(ROWS1M) check("one inverter: no stop at any edge crossed while charging", len({q(r["now"]) for r in ROWS1N if r["action"] == arbitrage.CHARGE}) >= 5 and not stops(ROWS1N)) check("non-vacuous: with the old projection one inverter stops at every edge " "for want of a measured headroom (5a4cdc2 then latched the quarter; " "the latch fix alone keeps it to that tick)", len(_m1) >= 2 and all("headroom" in r["reason"] and r["now"].astimezone(L).minute % 15 == 0 and r["now"].astimezone(L).second == 0 for _, r in _m1)) # Two inverters: since SAFETY-91 the guard grades the next quarter past a # block's end (`tariff.next_quarter()`), so the old projection no longer # stops there; the rolled one gives the same quarters. ROWS1O, C1O = run(1.9, 1.6, lag=LAG, patch=_no_roll, choice=_CH1) print(f" two inverters, old projection: {len(stops(ROWS1O))} stops; banked " f"{[round(x) for x in banked(C1O, ROWS1O)]} against " f"{[round(x) for x in banked(C1, ROWS1)]} rolled") check("two inverters: SAFETY-91 already holds the edge (no stop with the old " "projection), and rolling it changes no banked quarter", not stops(ROWS1O) and banked(C1O, ROWS1O) == banked(C1, ROWS1)) # 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) # 1e. Review of #157: gaps cannot flip the charge. A headroom forced to None # every other cycle flipped it 1799 times in 30 min before this; the second # unmeasured stop in a quarter now latches it. def _gaps(c): real, k = c._mode_roles, [0] def mr(now, maint): k[0] += 1 if k[0] % 2 == 0: c.arb_headroom_w = None return real(now, maint) c._mode_roles = mr _rows1e = run(2.0, 0.5, lag=LAG, patch=_gaps)[0] _flips = {} _on1e = next((i for i, r in enumerate(_rows1e) if r["action"] == arbitrage.CHARGE), len(_rows1e)) for _a, _b in zip(_rows1e[_on1e:], _rows1e[_on1e + 1:]): if (_a["action"] == arbitrage.CHARGE) != (_b["action"] == arbitrage.CHARGE): _flips[q(_b["now"])] = _flips.get(q(_b["now"]), 0) + 1 print(f" None every other cycle: charge on/off changes per quarter " f"{sorted(_flips.values())}") check("a headroom unmeasured every other cycle changes the charge at most " "4 times a quarter (restart, stop, restart, stop and latch)", _flips and max(_flips.values()) <= 4) # ============================================================================ # 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) # Review of #157: the hold's own rules, on `_reseed_run_edge()` and # `_edge_grid()` directly (target -10 W, deadband 15 W, EDGE_HOLD_S 18 s). _T0 = UTC0 def _held(grid, batt, s, last_b=-2500.0, end=False): """(what the loop reads, still held) for a charge end E=800 W at +s.""" f = SimpleNamespace( _edge=(800.0, 1.0, _T0 + timedelta(seconds=main.EDGE_HOLD_S), last_b), tuning=ctl.Tuning(target_grid_w=-10.0), grid=grid, site_batt_w=batt, i_w=0.0) g = main.Controller._edge_grid(f, end, _T0 + timedelta(seconds=s)) return g, f._edge is not None _hs = main.EDGE_HOLD_S check("the hold: the loop reads the target while the meter still shows the " "step; it ends on a new trade or claim, when the meter comes within the " "deadband of the target, and at EDGE_HOLD_S (later only while the " "battery still moves to E, and never past twice that)", _held(3300.0, -2500.0, 1) == (-10.0, True) and _held(3300.0, -2500.0, 1, end=True) == (3300.0, False) and _held(5.0, -2500.0, 1) == (-10.0, True) and _held(4.0, -2500.0, 1) == (4.0, False) and _held(3300.0, -2500.0, _hs + 1) == (3300.0, False) and _held(3300.0, -2000.0, _hs + 1) == (-10.0, True) and _held(3300.0, -2000.0, 2 * _hs + 1) == (3300.0, False)) # Review of #157 (X4): a reading that has passed E is not "still moving". check("...and a battery reading already past E does not extend the hold", _held(3300.0, 900.0, _hs + 1, last_b=700.0) == (3300.0, False)) _f = SimpleNamespace(site_batt_w=500.0, grid=90.0, _edge=None, tuning=ctl.Tuning(target_grid_w=-10.0), site_max_w=5000.0, loop_bound=lambda claims: None, target=None, i_w=None) main.Controller._reseed_run_edge(_f, [], _T0) check("...and a small step is held too (E 600 W from 500 W)", _f.target == 600.0 and _f._edge is not None and _f._edge[:2] == ( 600.0, 1.0)) # ============================================================================ # 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"])) # Review of #157 (M15): outside its probe a refuser is no charger, so the # others are not made to rest. With solar to spare, Solar first B takes it. _rows3s = run(1.9, 0.5, soc_a=30.0, refuse=("a",), steps=[(2, -1500)])[0] _pt = [r["now"] for r in _rows3s if r["probing"]] _sun = [r for r in _rows3s if r["refused"] and not r["probing"] and r["now"] >= at(2 + 5 / 60) and not any( timedelta(0) < r["now"] - t <= timedelta(seconds=30) for t in _pt)] _bsun = sorted(r["shares"].get(B, 0) for r in _sun) print(f" solar 1500 W over the house, A refusing: B's share outside probes " f"{_bsun[0] if _bsun else None} .. {_bsun[-1] if _bsun else None} W") check("...outside a probe the others are not held at rest: Solar first B " "takes the spare solar (from 30 s after a probe)", _sun and all(r["shares"].get(B, 0) < -1000 for r in _sun)) _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) # On the new base (release/1.0 e3e4ba6): SAFETY-78's drain comes before the # probe, and P4's day plan sets how far a probing Leftover solar battery goes. _ls = replace(_pk, mode=packmode.LEFTOVER_SOLAR) _dr = _dec([_pk, _pb], draining=True) check("a monthly drain comes before the probe: no grid charge while another " "battery drains, the probing refuser included (why 'cycle')", _dr.action is None and _dr.why == "cycle" and _dr.roles.get(A) != modetrade.CHARGE and _dec([_pk, _pb], draining=False).action == modetrade.CHARGE) check("with a day plan a probing Leftover solar battery is a charger up to " "its planned level and no further; without a level for it, it is not", _dec([_ls, _pb], targets={A: 60.0}).action == modetrade.CHARGE and _dec([_ls, _pb], targets={A: 25.0}).action is None and _dec([_ls, _pb], targets={B: 60.0}).action is None and _dec([_ls, _pb]).action is None and modetrade.energy([_ls], 20.0, 95.0, targets={A: 60.0})[1] == 0.0) # ============================================================================ # 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"(e3e4ba6: 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 e3e4ba6", _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 release/1.0 e3e4ba6") # Printed on release/1.0 e3e4ba6 (2026-10-13), goodwe-addon-checks:1, with this # branch's harness (same API). n1 and n2 are also 5a4cdc2's; the lag ones moved # with SAFETY-90/91 (n2_lag now equals n2: the next-quarter grading removed the # edge tick's blindness at N=2). # # PRICE-02 (2026-10-22) changed two things in the trade sentence and nothing # else (see test_modes_p3.py part 1): the "N above the cheapest" figure of a # sale and "hour" -> "price period". Three pins, as there: # AT_BASE_OLD the digests printed on e3e4ba6, sentence included (no longer # matched: the mask is needed); # AT_BASE the same rows through `ident.trade_digest()`, equal to what # e3e4ba6 gives through it; # AT_NOW this tree's own digests with nothing hidden. AT_BASE_OLD = { "n1": "71245b227ca40bada4f71f57ea6154475f4b0bd60239c9ad4f80bffb0094514c", "n2": "56cf28010b06765c29e5479abb815d7503d570b39240d82dfc818efa90c93722", "n1_lag": "86ef0f5ad8da1498ff7dcf59262a177bef847a0b201ce804e947c8eab5f07190", "n2_lag": "56cf28010b06765c29e5479abb815d7503d570b39240d82dfc818efa90c93722", } AT_BASE = { "n1": "33886ef03f364852883f0370230e6e04823fbb46a4704f88ca75bf9082d98c01", "n2": "666a31f41491c22750720524853d95944d5fba395d1d14040c460d71414b275f", "n1_lag": "f6c67b351bd732fb69e62cafbfdb2bcff6032f5d8845836659d096b8c677d3c0", "n2_lag": "666a31f41491c22750720524853d95944d5fba395d1d14040c460d71414b275f", } AT_NOW = { "n1": "939b3d47bbd64fa50e3d70538f092fb98e5c4bd5ead85a7fc696a1dbb514039e", "n2": "c51008213dd739e1e5142c236d9a950aa585992393e4af120086f1ba52ff3e0d", "n1_lag": "78aac892975be81a7ceac0c000ed441a731bd4f42ff9f729996cbee53d6b7f02", "n2_lag": "c51008213dd739e1e5142c236d9a950aa585992393e4af120086f1ba52ff3e0d", } _cmd_digest = ident.trade_digest _rows5 = {k: ident.run(**ident.SCENARIOS[k])[0] for k in AT_BASE} _got = {k: _cmd_digest(v) for k, v in _rows5.items()} _full5 = {k: ident.digest(v) for k, v in _rows5.items()} for _k in AT_BASE: print(f" {_k} {'same' if _got[_k] == AT_BASE[_k] else 'DIFFERS'} " f"{_got[_k][:16]}") check("with battery modes off every command, integrator, allocation and " "trade action equals e3e4ba6's (the sentence masked as above), also when " "every quarter boundary's cycle still sees the old P1 block", _got == AT_BASE) check("...the mask is needed and nothing more: unmasked, the digests differ " "from e3e4ba6's printed ones, and equal this tree's own pins, so a " "later change to any sentence or figure fails here", all(_full5[k] != AT_BASE_OLD[k] for k in AT_BASE_OLD) and _full5 == AT_NOW) _real_project = tariff.PeakTracker.project tariff.PeakTracker.project = staticmethod( lambda a, now, roll=False: _real_project(a, now, roll=True)) try: _moved = _cmd_digest(ident.run(**ident.SCENARIOS["n1_lag"])[0]) \ != AT_BASE["n1_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")