OPT-20: a refusal probe asks only PROBE_W, not the whole trade

The probe's trade was sized at the probing battery's rating. Its SAFETY-73
pin held 200 W of that, and the allocator put the rest on a battery with no
mode range (one in a maintenance phase): TEST-23 OPT-15 g7, -2300 W on A in
its maintenance charge for 34 s, grid 3100 W.

modetrade.decide() now asks PROBE_W for the probing battery plus what the
other chargers would have had without the probe; size() adds the probe on top
of their share, or files PROBE_W exactly, or nothing. _mode_trade() carries
the pin only on the ticks it is filed, so the window's first and last cycle
do not move another battery's share either. Modes off and modes on with no
probe are unchanged (bf9019e digests). Tests: test_opt20.py.

Co-Authored-By: Claude <noreply@anthropic.com>

goodwe_controller/tools/opt20_scen.py

@@ -0,0 +1,160 @@
+"""OPT-20: the refusal-probe scenarios, on the REAL Controller in the lagged
+replay (tools/opt11p3_replay.py). Uses only the API bf9019e has, so the same
+file runs on both trees (test_opt20.py pins the digests printed on bf9019e).
+
+Battery B is Cheap hours and refuses to charge; it is probed every 15 minutes
+inside the cheap run (OPT-15). A is, per scenario, in its maintenance charge
+phase, idle (Solar first), or a normal charger in the same run.
+"""
+
+import hashlib
+import logging
+import os
+import sys
+from datetime import datetime, timedelta, timezone
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+sys.path.insert(0, os.path.dirname(HERE))
+logging.disable(logging.WARNING)
+
+import opt11p3_replay as rp                                    # noqa: E402
+from app import packmode                                       # noqa: E402
+from app.maintenance import CHARGE, DRAIN, MaintResult         # noqa: E402
+
+L = timezone(timedelta(hours=2))
+A, B = "inverter-a", "inverter-b"
+LAG = rp.Lag(tau_s=0.05, ramp_w_s=1e9)
+SF_CH = {"a": (True, packmode.SOLAR_FIRST, False),
+         "b": (True, packmode.CHEAP_HOURS, False)}
+CH_CH = {"a": (True, packmode.CHEAP_HOURS, False),
+         "b": (True, packmode.CHEAP_HOURS, False)}
+
+# A day of the owner's kind: cheap 02:00-04:00 local, flat otherwise.
+D0 = datetime(2026, 9, 9, 0, 0, tzinfo=L)
+UTC0 = D0.astimezone(timezone.utc)
+_PR = [0.30] * 192
+for _i in range(8, 16):
+    _PR[_i] = 0.10
+WPTS = [(UTC0 + timedelta(minutes=15 * i), 900, p) for i, p in enumerate(_PR)]
+
+# OPT-15 rig g7: prices 0 from 17:45 to 18:45 local, flat 0.06 otherwise.
+G0 = datetime(2026, 10, 13, 0, 0, tzinfo=L)
+GUTC0 = G0.astimezone(timezone.utc)
+_GP = [0.06] * 192
+for _i in range(71, 75):
+    _GP[_i] = 0.0
+for _i in range(76, 78):
+    _GP[_i] = 0.5
+G7PTS = [(GUTC0 + timedelta(minutes=15 * i), 900, p)
+         for i, p in enumerate(_GP)]
+
+
+def at(h, base=UTC0):
+    return base + timedelta(hours=h)
+
+
+def _rows(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 "",
+            probing=c.inverters[1].probe_from is not None,
+            roles=dict(c.mode_tick.roles) if c.mode_tick else {},
+            p={u: p.p for u, p in h.plants.items()}))
+    return cb
+
+
+def day(choice, *, a_maint=None, shaped=False, refuse=("b",), hours=0.5,
+        house_w=800.0):
+    """02:00..: a cheap run, `house_w` house. `a_maint` is A's forced maintenance
+    result from 5 minutes in (B's refusal stands by then); `shaped` closes the
+    maintenance charge's own watts (charge-only shaping, hi = 0)."""
+    rows = []
+
+    def patch(c):
+        if shaped:
+            c._charge_headroom = lambda floor, claims: 0.0
+        if a_maint is None:
+            return
+        m, real = c.maints[0], c.maints[0].tick
+
+        def tick(now, soc, blocked_by=None, gate_open=True):
+            if now >= at(1.9 + 5 / 60):
+                return a_maint
+            return real(now, soc, blocked_by=blocked_by, gate_open=gate_open)
+        m.tick = tick
+    rp.run("p3_ch_sell", [(UTC0.timestamp() - 10, house_w)], WPTS, at(1.9),
+           3600 * hours, check=_rows(rows), lag=LAG, n=2,
+           soc0={"a": 50.0, "b": 30.0}, choice=choice, refuse=refuse,
+           patch=patch, extra={"cycle_cost_threshold_eur": 0.10})
+    return rows
+
+
+def g7():
+    """The OPT-15 rig's g7: B refuses from the start of the cheap run, A
+    drains 18:00 to 18:17:04 and then sits in its maintenance charge phase
+    (charge-only) at 0 W. The probe after the drain used to put the trade's
+    2500 W on A."""
+    rows = []
+    t_end = at(18 + 17 / 60 + 4 / 3600, GUTC0)
+
+    def patch(c):
+        m, real = c.maints[0], c.maints[0].tick
+
+        def tick(now, soc, blocked_by=None, gate_open=True):
+            if at(18.0, GUTC0) <= now < t_end:
+                return MaintResult(DRAIN, True, 2000.0, False, [])
+            if now >= t_end:
+                return MaintResult(CHARGE, False, None, True, [])
+            return real(now, soc, blocked_by=blocked_by, gate_open=gate_open)
+        m.tick = tick
+    rp.run("p3_ch_sell", [(GUTC0.timestamp() - 10, 800.0)], G7PTS,
+           at(17.6, GUTC0), 3600 * 1.1, check=_rows(rows), lag=LAG, n=2,
+           soc0={"a": 17.0, "b": 73.0}, choice=SF_CH, refuse=("b",),
+           patch=patch, extra={"cycle_cost_threshold_eur": 0.10})
+    return rows
+
+
+MAINT_CHARGE = MaintResult(CHARGE, False, None, True, [])
+
+
+def windows(rows):
+    """[(first, last)] row indexes of each probe window."""
+    out, i = [], 0
+    while i < len(rows):
+        if rows[i]["probing"]:
+            j = i
+            while j + 1 < len(rows) and rows[j + 1]["probing"]:
+                j += 1
+            out.append((i, j))
+            i = j + 1
+        else:
+            i += 1
+    return out
+
+
+def digest(rows) -> str:
+    keep = [(r["now"].isoformat(), sorted(r["shares"].items()),
+             round(r["grid"], 3), r["action"], r["reason"],
+             sorted(r["roles"].items()), sorted(r["p"].items()))
+            for r in rows]
+    return hashlib.sha256(repr(keep).encode()).hexdigest()
+
+
+def no_probe_digests() -> dict:
+    """Battery modes on, no refusal, so no probe: a scripted cheap run."""
+    return {
+        "sf_ch": digest(day(SF_CH, refuse=(), hours=1.2)),
+        "ch_ch": digest(day(CH_CH, refuse=(), hours=1.2)),
+        "ch_ch_maint": digest(day(CH_CH, a_maint=MAINT_CHARGE, refuse=(),
+                                  hours=1.2)),
+    }
+
+
+if __name__ == "__main__":
+    for k, v in no_probe_digests().items():
+        print(k, v)