glenn schrooyenandClaude
382b0ea411
With battery modes on, the price gap a mode needs was always widened x1.5 by Uncertainty.NO_WEATHER, even with a solar forecast read, so a 0.08 day/night gap never traded at the shipped cycle cost. The doubt is now added only when no usable forecast is read (solar_fc is None, the day plan's own test), in the run plan (_price_runs), the charge gate (_mode_roles) and Home's mirror (homeview.trade_verdict). The rule is one helper, arbitrage.weather_uncertainty(); the modes-off trade path (main.py cycle) keeps its literal NO_WEATHER. DOCS "Battery modes" sentence, CHANGELOG 0.19.1 line and the DECISIONS note follow. Tests: test_tariff24_weather.py. Co-Authored-By: Claude <noreply@anthropic.com>
355 lines
15 KiB
Python
355 lines
15 KiB
Python
"""Runnable check for TARIFF-24 round 2 (AUDIT-02 F1). `python3 test_tariff24_weather.py`
|
|
|
|
With battery modes on, the price gap a mode needs was always widened x1.5
|
|
(`Uncertainty.NO_WEATHER`) even with a solar forecast read. Now the doubt is
|
|
added only when no usable forecast is read (`arbitrage.weather_uncertainty()`),
|
|
in the run plan (`_price_runs()`), the charge gate (`_mode_roles()`) and Home's
|
|
mirror (`homeview.trade_verdict()`). Modes off is untouched.
|
|
DECISIONS/TARIFF-24-modes-need-the-tariff-task.md.
|
|
|
|
1. The rule itself, and that modes off keeps the doubt.
|
|
2. Day/night 0.34/0.26 (gap 0.08), modes on, real Controller, night period:
|
|
with a forecast at cycle cost 0.07 the cheap run is planned and the Cheap
|
|
hours battery charges; without one, or with a stale one, the same cost needs
|
|
0.105 and the reason is `not_worth` / "Prices differ too little".
|
|
3. A dynamic day with a 0.12 gap at cost 0.10: on with a forecast, not worth
|
|
without.
|
|
4. The forecast going and coming mid-run: the gate follows in the same cycle
|
|
the forecast read changes, both ways.
|
|
5. Home agrees with control in each case (Batteries Why, Home line, and
|
|
`trade_verdict()`, Home's mirror of the gate).
|
|
|
|
Mutations (MUTATION.md, purged copies, graded over this file): the doubt always
|
|
added (the old code) fails parts 2 to 5; never added fails the without-forecast
|
|
rows; Home's mirror left on the old rule fails part 5.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import math
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
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.join(HERE, "tools"))
|
|
logging.disable(logging.WARNING)
|
|
|
|
import opt11p3_replay as rp # noqa: E402
|
|
from named_absence import first, nth, present # noqa: E402
|
|
import app.main as m # noqa: E402
|
|
from app import arbitrage, homeview, modetrade # noqa: E402
|
|
from app.main import Controller # noqa: E402
|
|
from app.p1 import make_sample, split_signed # noqa: E402
|
|
from app.price import (PricePoint, PriceProduct, PriceRead, # noqa: E402
|
|
Uncertainty, brussels)
|
|
from app.static_tariff import TariffPlan # noqa: E402
|
|
from app.store import unit_key # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
A, B = "inverter-a", "inverter-b"
|
|
PROFILE = ('{"periods": ['
|
|
'{"name": "night", "days": "all", "from": "22:00", "to": "07:00",'
|
|
' "eur_per_kwh": 0.26},'
|
|
'{"name": "day", "days": "all", "from": "07:00", "to": "22:00",'
|
|
' "eur_per_kwh": 0.34}]}')
|
|
L = timezone(timedelta(hours=1))
|
|
NIGHT = datetime(2026, 10, 30, 23, 0, tzinfo=L).astimezone(timezone.utc)
|
|
CHOICE = {"a": (True, "leftover_solar", False),
|
|
"b": (True, "cheap_hours", False)}
|
|
FC_TODAY, FC_TOMORROW = "sensor.fc_today", "sensor.fc_tomorrow"
|
|
|
|
|
|
def opts(cost, *, arb=False, sensors=True):
|
|
o = dict(rp.options(True, arb, 2))
|
|
o.update(cycle_cost_threshold_eur=cost, static_profile=PROFILE)
|
|
if sensors:
|
|
o.update(pv_forecast_entity=FC_TODAY,
|
|
pv_forecast_tomorrow_entity=FC_TOMORROW)
|
|
return o
|
|
|
|
|
|
def dynamic_points(start):
|
|
"""A dynamic day-ahead shape, 0.16 .. 0.28 (gap 0.12), 15-minute points."""
|
|
out, t = [], start.replace(minute=0, second=0, microsecond=0)
|
|
for _ in range(48 * 4):
|
|
h = brussels(t).hour + brussels(t).minute / 60
|
|
p = 0.22 + 0.06 * math.cos((h - 19) / 24 * 2 * math.pi)
|
|
out.append(PricePoint(start=brussels(t), eur_per_kwh=round(p, 4),
|
|
provisional=False))
|
|
t += timedelta(minutes=15)
|
|
return out
|
|
|
|
|
|
def sensor(kwh, at):
|
|
return {"state": f"{kwh}", "attributes": {"unit_of_measurement": "kWh"},
|
|
"last_updated": at.isoformat(), "last_reported": at.isoformat()}
|
|
|
|
|
|
def forecast_mode(mode, t, now):
|
|
"""What the two forecast sensors read: 'fresh', 'stale', 'gone', or a
|
|
function of the simulated seconds that returns one of those."""
|
|
if callable(mode):
|
|
mode = mode(t)
|
|
if mode == "gone":
|
|
return {"state": "unavailable", "attributes": {}}
|
|
at = now if mode == "fresh" else now - timedelta(hours=5)
|
|
return at
|
|
|
|
|
|
def drive(o, *, fc="fresh", points=None, minutes=20, tick=5.0,
|
|
house_w=300.0, soc0=None):
|
|
"""-> (controller, rows). `points`: a feed-shaped plan injected every
|
|
cycle (no task); else the static profile via the real tariff task."""
|
|
store = rp._Store()
|
|
for u, (use, mode, sell) in CHOICE.items():
|
|
for k, val in (("pack_use", use), ("pack_mode", mode),
|
|
("pack_sell", sell)):
|
|
store.data[unit_key(f"inverter-{u}", k)] = val
|
|
rows = []
|
|
|
|
class H(rp.Hass):
|
|
async def state(self, e):
|
|
now = NIGHT + timedelta(seconds=self.t)
|
|
if fc is None:
|
|
return None
|
|
at = forecast_mode(fc, self.t, now)
|
|
if isinstance(at, dict):
|
|
return at
|
|
return sensor(0.0 if e == FC_TODAY else 12.0, at)
|
|
|
|
async def go():
|
|
hass = H(("a", "b"), lambda t: house_w, soc0 or {"a": 40.0, "b": 40.0})
|
|
tmp = tempfile.mkdtemp(prefix="t24w-")
|
|
c = Controller(dict(o), hass, store, rp._Pub(), stop=asyncio.Event(),
|
|
peak_db_path=os.path.join(tmp, "tariff.db"),
|
|
price_history_path=os.path.join(tmp, "ph.json"))
|
|
for i in c.inverters:
|
|
i.dev_min, i.dev_max = -5000.0, 5000.0
|
|
c.auto = True
|
|
real_dt, real_hv = m.datetime, homeview.datetime
|
|
clock = [NIGHT]
|
|
|
|
class _DT(real_dt):
|
|
@classmethod
|
|
def now(cls, tz=None):
|
|
return clock[0] if tz is None else clock[0].astimezone(tz)
|
|
|
|
m.datetime = homeview.datetime = _DT
|
|
pts = None if points is None else points
|
|
task = None
|
|
if pts is None:
|
|
task = asyncio.create_task(c.run_tariff_refresh())
|
|
try:
|
|
t = 0.0
|
|
while t < minutes * 60:
|
|
now = NIGHT + timedelta(seconds=t)
|
|
hass.t = t
|
|
clock[0] = now
|
|
if pts is not None:
|
|
c.tariff_plan = TariffPlan(
|
|
PriceRead.series(pts, PriceProduct.DAY_AHEAD,
|
|
"generic_dayahead", now), None,
|
|
"provider")
|
|
c.tariff_at = now
|
|
elif t == 0.0:
|
|
for _ in range(500):
|
|
if c.tariff_plan is not None:
|
|
break
|
|
await asyncio.sleep(0)
|
|
imp, exp = split_signed(hass.grid())
|
|
c.p1.submit(make_sample("ha_signed", imp, exp, phases=1,
|
|
ingest_ts=now,
|
|
ingest_mono=time.monotonic()))
|
|
await c.read_inputs(now_mono=t)
|
|
await c.cycle()
|
|
hass.advance(tick, 10.0)
|
|
rp_ = c.mode_runs
|
|
rows.append(dict(
|
|
t=t, now=now, fc_none=c.solar_fc is None,
|
|
runs_on=None if rp_ is None else rp_.on,
|
|
reason=None if rp_ is None else rp_.reason,
|
|
roles=dict(c.mode_tick.roles) if c.mode_tick else {},
|
|
why=None if c.mode_tick is None else c.mode_tick.why,
|
|
charging=hass.plants["b"].p < -200.0))
|
|
t += tick
|
|
c.home_plan = homeview.build_plan(c, clock[0], False, True)
|
|
c.cards = homeview.battery_rows(c, False, True)
|
|
# Home's mirror of the gate (it reads the trade switch, so ask it
|
|
# as a site that has it on next to the modes).
|
|
c2 = c
|
|
c2.o = dict(c.o, arbitrage_enabled=True)
|
|
c.mirror, c.mirror_why = homeview.trade_verdict(
|
|
c2, c.fresh_tariff_plan(clock[0]), clock[0])
|
|
c.o = dict(c.o, arbitrage_enabled=o.get("arbitrage_enabled",
|
|
False))
|
|
finally:
|
|
if task is not None:
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
m.datetime, homeview.datetime = real_dt, real_hv
|
|
return c
|
|
|
|
return asyncio.run(go()), rows
|
|
|
|
|
|
def why_of(c, name):
|
|
card = first((b for b in c.cards if b["name"] == name), f"card {name}")
|
|
return card["why_key"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("1. the rule")
|
|
NW = frozenset({Uncertainty.NO_WEATHER})
|
|
check("modes on with a forecast read: no doubt", arbitrage.weather_uncertainty(
|
|
True, object()) == frozenset())
|
|
check("modes on with no forecast (None): the doubt", arbitrage.weather_uncertainty(
|
|
True, None) == NW)
|
|
check("modes off, forecast or not: the doubt, as before",
|
|
arbitrage.weather_uncertainty(False, object()) == NW
|
|
and arbitrage.weather_uncertainty(False, None) == NW)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n2. day/night 0.34 / 0.26 (gap 0.08), night period")
|
|
c1, r1 = drive(opts(0.07), fc="fresh")
|
|
l1 = nth(r1, -1, "last cycle")
|
|
print(f" forecast read, cost 0.07: mode_runs on={l1['runs_on']} "
|
|
f"reason={l1['reason']}; B charging {sum(r['charging'] for r in r1)} of "
|
|
f"{len(r1)}; Why A {why_of(c1, A)} B {why_of(c1, B)}")
|
|
check("with a forecast at cost 0.07 the cheap run is planned (needed 0.07 < "
|
|
"0.08) and the Cheap hours battery charges in the night period",
|
|
present(l1) and not l1["fc_none"] and l1["runs_on"] is True
|
|
and sum(r["charging"] for r in r1) > len(r1) // 2
|
|
and l1["roles"].get(B) == modetrade.CHARGE)
|
|
for label, mode in (("no forecast sensor reads", "gone"),
|
|
("a stale forecast", "stale")):
|
|
c, rr = drive(opts(0.07), fc=mode)
|
|
ll = nth(rr, -1, "last cycle")
|
|
print(f" {label}, cost 0.07: on={ll['runs_on']} reason={ll['reason']}; "
|
|
f"B charging {sum(r['charging'] for r in rr)}; Why A "
|
|
f"{why_of(c, A)} B {why_of(c, B)}")
|
|
check(f"{label}: the same cost needs 0.105 > 0.08, so not_worth and "
|
|
f"'Prices differ too little', no grid charge",
|
|
present(ll) and ll["fc_none"] and ll["runs_on"] is False
|
|
and ll["reason"] == "not_worth"
|
|
and why_of(c, A) == "no_spread" and why_of(c, B) == "no_spread"
|
|
and not any(r["charging"] for r in rr))
|
|
c0, r0 = drive(opts(0.07, sensors=False), fc=None)
|
|
check("no forecast sensor configured at all: not_worth too (as 0.19.0)",
|
|
present(nth(r0, -1)) and r0[-1]["runs_on"] is False)
|
|
c2, r2 = drive(opts(0.10), fc="fresh")
|
|
check("at the shipped cost 0.10 even a forecast does not make 0.08 enough "
|
|
"(needs 0.10): still not_worth, honestly",
|
|
present(nth(r2, -1)) and r2[-1]["runs_on"] is False
|
|
and r2[-1]["reason"] == "not_worth")
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n3. a dynamic day, gap 0.12, cost 0.10")
|
|
pts = dynamic_points(NIGHT - timedelta(hours=1))
|
|
cd, rd = drive(opts(0.10), fc="fresh", points=pts)
|
|
cn, rn = drive(opts(0.10), fc="gone", points=pts)
|
|
_ld, _ln = nth(rd, -1, "last cycle"), nth(rn, -1, "last cycle")
|
|
print(f" with forecast: on={_ld['runs_on']} reason={_ld['reason']}; "
|
|
f"without: on={_ln['runs_on']} reason={_ln['reason']}")
|
|
check("with a forecast the run plan is on (needs 0.10 < 0.12)",
|
|
present(nth(rd, -1)) and rd[-1]["runs_on"] is True
|
|
and why_of(cd, A) != "no_spread")
|
|
check("without one it is not_worth (needs 0.15 > 0.12) and says so",
|
|
present(nth(rn, -1)) and rn[-1]["runs_on"] is False
|
|
and rn[-1]["reason"] == "not_worth"
|
|
and why_of(cn, B) == "no_spread")
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n4. the forecast going and coming mid-run")
|
|
CUT = 600.0 # 10 min in
|
|
|
|
|
|
def going(t):
|
|
return "fresh" if t < CUT else "gone"
|
|
|
|
|
|
def coming(t):
|
|
return "gone" if t < CUT else "fresh"
|
|
|
|
|
|
cg, rg = drive(opts(0.07), fc=going, minutes=25)
|
|
cc, rc = drive(opts(0.07), fc=coming, minutes=25)
|
|
|
|
|
|
def flips(rows):
|
|
"""Cycles where the forecast read changed, and whether the plan and the
|
|
charge gate changed in that very cycle."""
|
|
out = []
|
|
for i in range(1, len(rows)):
|
|
if rows[i]["fc_none"] != rows[i - 1]["fc_none"]:
|
|
out.append((i, rows[i]["t"], rows[i]["runs_on"],
|
|
rows[i]["charging"], rows[i - 1]["charging"]))
|
|
return out
|
|
|
|
|
|
fg, fcm = flips(rg), flips(rc)
|
|
print(f" going: forecast read lost at cycle/t {fg}; coming: {fcm}")
|
|
check("going: the plan is on with the forecast and not_worth the cycle the "
|
|
"read is lost (within one forecast read, 60 s, of the sensor change)",
|
|
len(fg) == 1 and fg[0][1] - CUT <= 60.0 and fg[0][2] is False
|
|
and all(r["runs_on"] is True for r in rg if not r["fc_none"])
|
|
and all(r["runs_on"] is False for r in rg if r["fc_none"]))
|
|
check("...and the grid charge stops with it: no charging after the cycle "
|
|
"that lost the forecast, plus the 15 s the inverter ramp takes",
|
|
len(fg) == 1 and not any(
|
|
r["charging"] for r in rg if r["t"] >= fg[0][1] + 40.0))
|
|
check("coming: not_worth until the read is back, then on in the same cycle, "
|
|
"and the battery charges",
|
|
len(fcm) == 1 and fcm[0][1] - CUT <= 60.0 and fcm[0][2] is True
|
|
and all(r["runs_on"] is False for r in rc if r["fc_none"])
|
|
and all(r["runs_on"] is True for r in rc if not r["fc_none"])
|
|
and any(r["charging"] for r in rc if r["t"] >= fcm[0][1] + 40.0))
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n5. Home agrees with control")
|
|
CASES = (("forecast, cost 0.07", c1, r1, True),
|
|
("no forecast, cost 0.07", *drive(opts(0.07), fc="gone"), False),
|
|
("dynamic, forecast", cd, rd, True),
|
|
("dynamic, no forecast", cn, rn, False))
|
|
for label, cx, rx, on in CASES:
|
|
last = nth(rx, -1, "last cycle")
|
|
hp = cx.home_plan
|
|
mirror_ok = cx.mirror is not None
|
|
print(f" {label}: control on={last['runs_on']}; Home mirror "
|
|
f"{'allows' if mirror_ok else 'refuses'} ({cx.mirror_why}); "
|
|
f"lines {hp['lines'][:1]}; Why B {why_of(cx, B)}")
|
|
check(f"{label}: Home's mirror of the gate allows exactly when control "
|
|
f"plans", present(last) and mirror_ok is on
|
|
and last["runs_on"] is on)
|
|
check(f"{label}: the Batteries screen and the Home line say the gap is "
|
|
f"too small exactly when control does",
|
|
(why_of(cx, B) == "no_spread") is (not on)
|
|
and any("price gap is too small" in ln for ln in hp["lines"])
|
|
is (not on))
|
|
if label.startswith("dynamic"):
|
|
continue # its cheap run is tomorrow morning: not on today's chart
|
|
check(f"{label}: Home draws a grid charge block exactly when control "
|
|
f"charges", any(r["kind"] == "grid" for r in hp["runs"]) is on)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED:")
|
|
for f in fails:
|
|
print(f" - {f}")
|
|
sys.exit(1)
|
|
print("all checks passed")
|