glenn schrooyenandClaude
4297f7f3f1
TariffPlan.arbitrage(now=...) drops the points whose period has ended (end <= now), so the cheapest and dearest point can still be traded. decide() (modes off), the modes-on allowed() gate and the Home verdict all pass their own now. Hourly feeds and the static tariff are unchanged. The off-peak sentence and the charge/discharge sentences say "price period". Tests: test_price02.py; test_arbitrage.py rolls its whole-day plans to start at now; test_modes_p3.py part 1 digests leave the trade sentence out (commands unchanged on7a00f12,90bc901and this tree). Co-Authored-By: Claude <noreply@anthropic.com>
322 lines
14 KiB
Python
322 lines
14 KiB
Python
"""Runnable check for PRICE-02. `python3 test_price02.py`
|
|
|
|
The trading verdict is taken over the price periods that have not ended.
|
|
`TariffPlan.arbitrage(now=...)` drops the points with `end <= now` (the span
|
|
`price._cover_table()` gives each one) and every caller passes its `now`:
|
|
`arbitrage.decide()` (modes off), the `allowed()` gate in
|
|
`Controller._mode_roles()` (modes on) and the Home page's `trade_verdict()`.
|
|
|
|
1. A 15-minute day, probed at :05, :20, :35 and :50 of the cheap hour: the
|
|
buy leg and the sell leg move to the current period and never name one that
|
|
has ended, and `decide()` trades in it. Cheapest in the past, the current
|
|
and a future period are all covered.
|
|
2. A plan built at 03:00 and used at 03:20 (the 30-minute refresh lag), and the
|
|
shipped refresh (`fill_from_yesterday`, which keeps the whole current hour).
|
|
3. An hourly feed and the static day/night tariff are unchanged: an hourly
|
|
point that is still current stays in.
|
|
4. Modes on: a spread that exists only because of an ended period no longer
|
|
opens the gate (`Controller._mode_roles`), nor the Home page verdict.
|
|
5. The reason text says "price period".
|
|
|
|
Mutations: MUTATION.md says how to run them (purged copies); the ones of
|
|
PRICE-02 are listed in the pull request.
|
|
"""
|
|
|
|
import json
|
|
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.join(HERE, "tools"))
|
|
logging.disable(logging.WARNING)
|
|
|
|
from app import arbitrage, homeview # noqa: E402
|
|
from app.arbitrage import CHARGE, DISCHARGE, decide # noqa: E402
|
|
from app.price import (PricePoint, PriceProduct, PriceRead, # noqa: E402
|
|
Uncertainty, brussels)
|
|
from app.price_forecast import fill_from_yesterday # noqa: E402
|
|
from app.static_tariff import TariffPlan, static_plan # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
UTC = timezone.utc
|
|
NO_WEATHER = frozenset({Uncertainty.NO_WEATHER})
|
|
D0 = datetime(2026, 10, 20, 0, 0, tzinfo=UTC) # 02:00 Brussels
|
|
H3 = D0 + timedelta(hours=1) # 03:00 Brussels
|
|
ARGS = dict(peak_guard_active=False, min_spread_eur_kwh=0.10,
|
|
uncertainty=NO_WEATHER)
|
|
|
|
|
|
def qplan(prices, start=D0, minutes=15, product=PriceProduct.DAY_AHEAD):
|
|
pts = [PricePoint(brussels(start + timedelta(minutes=minutes * i)), p,
|
|
False) for i, p in enumerate(prices)]
|
|
return TariffPlan(PriceRead.series(pts, product, "generic_dayahead",
|
|
start), None, "provider")
|
|
|
|
|
|
def dec(plan, now, **kw):
|
|
base = dict(peak_guard_active=False, charge_headroom_w=5000.0, soc=50.0,
|
|
min_spread_eur_kwh=0.10, uncertainty=frozenset(),
|
|
charge_w=2500.0, discharge_w=2500.0, soc_floor_pct=10.0,
|
|
soc_ceiling_pct=90.0)
|
|
base.update(kw)
|
|
return decide(plan, now, **base)
|
|
|
|
|
|
def hhmm(t):
|
|
return f"{brussels(t):%H:%M}"
|
|
|
|
|
|
def ahead_extreme(prices, now, pick, start=D0, minutes=15):
|
|
"""The expected leg, by plain arithmetic: the extreme price over the
|
|
periods whose end is after `now`, as the start instant (first on a tie)."""
|
|
best = None
|
|
for i, p in enumerate(prices):
|
|
s = start + timedelta(minutes=minutes * i)
|
|
if s + timedelta(minutes=minutes) <= now:
|
|
continue
|
|
if best is None or pick(p, best[1]):
|
|
best = (s, p)
|
|
return best[0]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("1. a 15-minute day: the verdict is taken from the current period on")
|
|
BUY = [0.30] * 192 # 48 h of quarters from 02:00
|
|
for q, p in ((4, 0.05), (5, 0.06), (6, 0.07), (7, 0.08), (68, 0.50)):
|
|
BUY[q] = p # 03:00 .05 03:15 .06 03:30 .07 03:45 .08 19:00 .50
|
|
BUY_PLAN = qplan(BUY)
|
|
check("the fixture: the day's cheapest period is 03:00, then three dearer "
|
|
"ones, and the dearest is 19:00 - so every probe has a past, a current "
|
|
"and a future period to choose between",
|
|
BUY.index(min(BUY)) == 4 and BUY.index(max(BUY)) == 68
|
|
and BUY[4] < BUY[5] < BUY[6] < BUY[7] < min(BUY[8:68]))
|
|
want_q = {5: 4, 20: 5, 35: 6, 50: 7}
|
|
for mins, q in want_q.items():
|
|
now = H3 + timedelta(minutes=mins)
|
|
v = BUY_PLAN.arbitrage(now=now, **ARGS)
|
|
d = dec(BUY_PLAN, now)
|
|
got = v.cheapest.start
|
|
exp = ahead_extreme(BUY, now, lambda a, b: a < b)
|
|
check(f"03:{mins:02d}: the buy leg is {hhmm(exp)} (the first period not "
|
|
f"over), the verdict says {hhmm(got)}, and decide() is {d.action}",
|
|
got == exp == D0 + timedelta(minutes=15 * q)
|
|
and d.action == CHARGE
|
|
and abs(d.price_now_eur_kwh - BUY[q]) < 1e-9)
|
|
check("before the change the same plan at 03:20 aimed at 03:00 (the "
|
|
"verdict without `now`) and decide() did nothing - the defect, shown",
|
|
BUY_PLAN.arbitrage(**ARGS).cheapest.start == H3
|
|
and arbitrage._act(BUY_PLAN, H3 + timedelta(minutes=20),
|
|
BUY_PLAN.arbitrage(**ARGS), 50.0, 2500.0, 2500.0,
|
|
10.0, 90.0, 5000.0).action is None)
|
|
|
|
# A cheapest period still AHEAD stays the target from every probe.
|
|
FUT = [0.30] * 192
|
|
FUT[8], FUT[68] = 0.04, 0.50 # cheapest 04:00, dearest 19:00
|
|
FUT_PLAN = qplan(FUT)
|
|
check("the cheapest in the FUTURE (04:00) is the buy leg at :05, :20, :35 and "
|
|
":50 of the hour before it - nothing ahead is dropped",
|
|
all(FUT_PLAN.arbitrage(now=H3 + timedelta(minutes=m), **ARGS)
|
|
.cheapest.start == H3 + timedelta(hours=1) for m in (5, 20, 35, 50)))
|
|
check("...and at 04:20 (that period is now the current one) it is still the "
|
|
"current period at :00-:15 only: 04:20 aims elsewhere",
|
|
FUT_PLAN.arbitrage(now=H3 + timedelta(hours=1, minutes=20), **ARGS)
|
|
.cheapest.start != H3 + timedelta(hours=1))
|
|
|
|
print("\n1b. the sell leg")
|
|
SELL = [0.30] * 192
|
|
for q, p in ((4, 0.70), (5, 0.65), (6, 0.60), (7, 0.55), (68, 0.10)):
|
|
SELL[q] = p # dearest 03:00 .70 .. 03:45 .55 ; cheapest 19:00
|
|
SELL_PLAN = qplan(SELL)
|
|
for mins, q in want_q.items():
|
|
now = H3 + timedelta(minutes=mins)
|
|
v = SELL_PLAN.arbitrage(now=now, **ARGS)
|
|
d = dec(SELL_PLAN, now)
|
|
exp = ahead_extreme(SELL, now, lambda a, b: a > b)
|
|
check(f"03:{mins:02d}: the sell leg is {hhmm(exp)}, the verdict says "
|
|
f"{hhmm(v.dearest.start)}, and decide() is {d.action}",
|
|
v.dearest.start == exp == D0 + timedelta(minutes=15 * q)
|
|
and d.action == DISCHARGE
|
|
and abs(d.price_now_eur_kwh - SELL[q]) < 1e-9)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n2. the plan is older than the verdict")
|
|
# Built at 03:00 and used at 03:20, 03:35 (the refresh runs every 30 min).
|
|
built = H3
|
|
feed = qplan(BUY).read
|
|
plan_0300 = TariffPlan(fill_from_yesterday(
|
|
feed, yesterday=PriceRead.no_data("none"), since=built,
|
|
until=built + timedelta(hours=48), stamp=built), None, "provider")
|
|
check("the plan built at 03:00 starts at 03:00 (the refresh floors to the hour)",
|
|
min(p.start for p in plan_0300.read.points) == H3)
|
|
for mins, q in ((20, 5), (35, 6)):
|
|
now = H3 + timedelta(minutes=mins)
|
|
d = dec(plan_0300, now)
|
|
check(f"used at 03:{mins:02d} the verdict is "
|
|
f"{hhmm(D0 + timedelta(minutes=15 * q))} and decide() charges there",
|
|
plan_0300.arbitrage(now=now, **ARGS).cheapest.start
|
|
== D0 + timedelta(minutes=15 * q) and d.action == CHARGE)
|
|
# The shipped refresh at 03:20 keeps the whole hour: the verdict still skips it.
|
|
now = H3 + timedelta(minutes=20)
|
|
plan_refresh = TariffPlan(fill_from_yesterday(
|
|
feed, yesterday=PriceRead.no_data("none"), since=now,
|
|
until=now + timedelta(hours=48), stamp=now), None, "provider")
|
|
check("the shipped refresh at 03:20 still holds the 03:00 period, as before, "
|
|
"and the verdict does not aim at it",
|
|
min(p.start for p in plan_refresh.read.points) == H3
|
|
and plan_refresh.arbitrage(now=now, **ARGS).cheapest.start
|
|
== H3 + timedelta(minutes=15))
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n3. hourly feeds and the static tariff do not move")
|
|
HR = [0.30] * 48 # 48 hourly points from 02:00
|
|
HR[1], HR[17] = 0.05, 0.50 # cheapest 03:00, dearest 19:00
|
|
HR_PLAN = qplan(HR, minutes=60)
|
|
same = True
|
|
for m in (0, 5, 20, 35, 59):
|
|
now = H3 + timedelta(minutes=m)
|
|
a = HR_PLAN.arbitrage(now=now, **ARGS)
|
|
b = HR_PLAN.arbitrage(**ARGS)
|
|
same = same and a.cheapest is b.cheapest and a.dearest is b.dearest \
|
|
and a.spread_eur_kwh == b.spread_eur_kwh
|
|
check("an hourly point that is still current stays in: at 03:00, :05, :20, "
|
|
":35 and :59 the verdict is the one without `now`", same)
|
|
check("...and decide() still charges in the current hour at 03:40",
|
|
dec(HR_PLAN, H3 + timedelta(minutes=40)).action == CHARGE)
|
|
check("...and once the hour has ended its point is gone: at 04:05 the buy leg "
|
|
"is no longer 03:00",
|
|
HR_PLAN.arbitrage(now=H3 + timedelta(hours=1, minutes=5), **ARGS)
|
|
.cheapest.start != H3)
|
|
|
|
TWO = {"periods": [
|
|
{"name": "day", "from": "07:00", "to": "22:00", "days": "mon-sun",
|
|
"eur_per_kwh": 0.40},
|
|
{"name": "night", "from": "22:00", "to": "07:00", "days": "mon-sun",
|
|
"eur_per_kwh": 0.18}]}
|
|
OPTS = {"price_provider": "off", "static_profile": json.dumps(TWO)}
|
|
ok, n_checked = True, 0
|
|
for h in range(0, 24):
|
|
for m in (0, 20, 50):
|
|
now = D0 + timedelta(hours=h - 2, minutes=m) # from 22:00 UTC the day before
|
|
plan = static_plan(OPTS, now, now + timedelta(hours=48), now=now)
|
|
a = plan.arbitrage(now=now, **ARGS)
|
|
b = plan.arbitrage(**ARGS)
|
|
n_checked += 1
|
|
ok = ok and a.kind == b.kind == "allowed" and a.cheapest is b.cheapest \
|
|
and a.dearest is b.dearest and a.spread_eur_kwh == b.spread_eur_kwh
|
|
check(f"the static day/night plan (built from the current hour, as main.py "
|
|
f"does) gives the same verdict with and without `now` at {n_checked} "
|
|
f"instants", ok and n_checked == 72)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n4. modes on: an ended period no longer opens the gate")
|
|
GATE = [0.30] * 192
|
|
GATE[4] = 0.10 # the ONLY spread is the 03:00 period
|
|
GATE_PLAN = qplan(GATE)
|
|
now = H3 + timedelta(minutes=20)
|
|
check("the fixture: with the ended 03:00 period in view the verdict is "
|
|
"allowed; from 03:20 on the rest of the series is flat",
|
|
GATE_PLAN.arbitrage(**ARGS).kind == "allowed"
|
|
and GATE_PLAN.arbitrage(now=now, **ARGS).kind == "refused"
|
|
and "flat" in GATE_PLAN.arbitrage(now=now, **ARGS).reason)
|
|
|
|
import opt11p3_replay as rp # noqa: E402
|
|
|
|
WPTS = [(D0 + timedelta(minutes=15 * i), 900, 0.10) for i in range(192)]
|
|
|
|
|
|
def controller_at(at, plan):
|
|
"""The real Controller after a few cycles of the modes-on replay at `at`
|
|
(two batteries, one on Cheap hours), holding `plan` as its fresh plan."""
|
|
_st, c = rp.run("p3_ch_sell", [(D0.timestamp() - 10, 800.0)], WPTS, at,
|
|
20.0, soc0={"a": 60.0, "b": 30.0},
|
|
extra={"cycle_cost_threshold_eur": 0.05})
|
|
c.tariff_plan, c.tariff_at = plan, at
|
|
return c
|
|
|
|
|
|
from app import modetrade # noqa: E402
|
|
|
|
GATE_SEEN = {}
|
|
_real_decide = modetrade.decide
|
|
|
|
|
|
def _spy(runs, now_, packs, **kw):
|
|
GATE_SEEN.update(charge_ok=kw["charge_ok"], sell_ok=kw["sell_ok"],
|
|
trade_why=kw["trade_why"])
|
|
return _real_decide(runs, now_, packs, **kw)
|
|
|
|
|
|
modetrade.decide = _spy
|
|
c = controller_at(now, GATE_PLAN)
|
|
GATE_SEEN.clear()
|
|
c._mode_roles(now, None)
|
|
late = dict(GATE_SEEN)
|
|
c2 = controller_at(H3 + timedelta(minutes=5), GATE_PLAN)
|
|
GATE_SEEN.clear()
|
|
c2._mode_roles(H3 + timedelta(minutes=5), None)
|
|
early = dict(GATE_SEEN)
|
|
modetrade.decide = _real_decide
|
|
check(f"Controller._mode_roles() (modes on, the `allowed()` gate) at 03:05, "
|
|
f"03:00 still current: charge and sell are allowed ({early})",
|
|
early.get("charge_ok") is True and early.get("sell_ok") is True)
|
|
check(f"...and at 03:20, with only flat periods ahead: neither is allowed, "
|
|
f"and the reason is the spread ({late})",
|
|
late.get("charge_ok") is False and late.get("sell_ok") is False
|
|
and late.get("trade_why") == "not_worth")
|
|
|
|
c.o["arbitrage_enabled"] = True
|
|
v, why = homeview.trade_verdict(c, GATE_PLAN, now)
|
|
check(f"the Home page verdict takes the same `now`: no verdict at 03:20 "
|
|
f"({why!r}), and a verdict at 03:05",
|
|
v is None and why == "not_worth"
|
|
and homeview.trade_verdict(c, GATE_PLAN, H3 + timedelta(minutes=5))[0]
|
|
is not None)
|
|
check("...and the refusal key counts only periods not over: two firm periods "
|
|
"left, so 'not_worth', and with one left 'not_firm'",
|
|
homeview.refusal_key(GATE_PLAN, False, now) == "not_worth"
|
|
and homeview.refusal_key(qplan([0.1, 0.2]), False,
|
|
D0 + timedelta(minutes=20)) == "not_firm")
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n5. the text")
|
|
check("the reason for an off-peak period says 'price period', not 'hour'",
|
|
arbitrage.IDLE_OFF_PEAK_HOUR
|
|
== "now is not the cheapest or the dearest price period")
|
|
_b = dec(BUY_PLAN, H3 + timedelta(minutes=20))
|
|
_s = dec(SELL_PLAN, H3 + timedelta(minutes=20))
|
|
check("...and the charge and the discharge sentences say 'price period' too "
|
|
"(the unit is 15 minutes), with no 'firm hour' left",
|
|
"cheapest firm price period" in _b.reason
|
|
and "dearest firm price period" in _s.reason
|
|
and "firm hour" not in _b.reason + _s.reason)
|
|
_hi = dec(BUY_PLAN, H3 + timedelta(minutes=20), soc=95.0)
|
|
_lo = dec(SELL_PLAN, H3 + timedelta(minutes=20), soc=10.0)
|
|
check("...and so do the full-pack and empty-pack sentences ('buy period', "
|
|
"'sell period')",
|
|
"the buy period is here" in _hi.reason
|
|
and "the sell period is here" in _lo.reason)
|
|
try:
|
|
BUY_PLAN.arbitrage(now=datetime(2026, 10, 20, 3, 20), **ARGS)
|
|
_naive = False
|
|
except TypeError as e:
|
|
_naive = "tz-aware" in str(e)
|
|
check("a naive `now` is refused: it is ambiguous for one hour every October",
|
|
_naive)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: " + "; ".join(fails))
|
|
sys.exit(1)
|
|
print("all checks passed")
|