glenn schrooyenandClaude
891ed7bdd7
- TariffPlan.arbitrage(now=...) is required (keyword-only, no default); so are homeview.trade_verdict/refusal_key's now. Existing test calls pass one. - test_arbitrage.py: no module-level decide wrapper. Fixtures that probed a one-day plan after its extremes now hold two days (as a site does); the flap day gets a unique cheapest and dearest hour; the replay breaks its ties without direction. The flap bound is <= 4 and the replay check is len(_trades) < 365 again. 228 checks pass on the old and the new code. - test_price02.py part 6: ties (equal quarters, a real 15-minute day, alternating hourly prices) and the required now. - Identity pins: the masked digest is only for the older-tree comparison; this tree's own digests are pinned with nothing hidden. - DECISIONS: the tie analysis with numbers. Co-Authored-By: Claude <noreply@anthropic.com>
665 lines
31 KiB
Python
665 lines
31 KiB
Python
"""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")
|