glenn schrooyenandClaude
3dd4dd89e0
Docs and tests only; no change under goodwe_controller/app. Co-Authored-By: Claude <noreply@anthropic.com> Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
539 lines
26 KiB
Python
539 lines
26 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: an hourly point that is
|
|
still current stays in. A plan built off the hour (the refresh runs every
|
|
30 min from add-on start, so :20 and :50 are normal) no longer aims at the
|
|
hour that has just ended, so a trade runs through (owner, 2026-10-22), and
|
|
a period that ends exactly at `now` is over.
|
|
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"; `now` is required (no default).
|
|
6. Ties. The verdict takes the first extreme still ahead, where it used to
|
|
take the first of the whole plan. On hour-constant 15-minute data (four
|
|
equal quarters) that is four acting periods in the cheap hour, not one;
|
|
the number of times the action changes does not grow. The same on a real
|
|
15-minute day, and hourly alternating prices change at most once an hour.
|
|
|
|
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
|
|
LONG_AGO = datetime(2000, 1, 1, tzinfo=UTC) # before every point: nothing has ended, the old reach
|
|
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")
|
|
|
|
|
|
ARGS_NO_NOW = dict(ARGS)
|
|
ARGS5 = dict(ARGS, min_spread_eur_kwh=0.05, uncertainty=frozenset())
|
|
|
|
|
|
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(now=LONG_AGO, **ARGS).cheapest.start == H3
|
|
and arbitrage._act(BUY_PLAN, H3 + timedelta(minutes=20),
|
|
BUY_PLAN.arbitrage(now=LONG_AGO, **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: a current hour stays in")
|
|
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(now=LONG_AGO, **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(now=LONG_AGO, **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 at the instant it is used (so it "
|
|
f"holds no ended hour), gives the same verdict with and without `now` "
|
|
f"at {n_checked} instants", ok and n_checked == 72)
|
|
|
|
# 3b. The refresh is NOT on the hour. main.py rebuilds the plan every 30 min
|
|
# counted from add-on start (TARIFF_REFRESH_S = 1800), so a plan built at :20
|
|
# or :50 starts at the floored hour and, until the next refresh, still holds
|
|
# the hour that has just ended. Owner decision 2026-10-22: the verdict skips
|
|
# that hour too, so a trade runs through instead of pausing until the refresh.
|
|
print("\n3b. hourly and static plans built off the hour: the ended hour is not the target")
|
|
HR2 = [0.30] * 48 # hourly from 02:00 Brussels
|
|
HR2[1], HR2[2], HR2[17] = 0.05, 0.05, 0.50 # cheap 03:00 and 04:00 (a tie), dear 19:00
|
|
built = H3 + timedelta(minutes=50) # the refresh at :50 floors to 03:00
|
|
plan_h = TariffPlan(fill_from_yesterday(
|
|
qplan(HR2, minutes=60).read, yesterday=PriceRead.no_data("none"),
|
|
since=built, until=built + timedelta(hours=48), stamp=built),
|
|
None, "provider")
|
|
now = H3 + timedelta(hours=1, minutes=5) # 04:05, before the next refresh
|
|
check("the hourly plan built at 03:50 starts at 03:00 and is used at 04:05, so "
|
|
"it still holds the 03:00 hour, which is over",
|
|
min(p.start for p in plan_h.read.points) == H3)
|
|
old_v = plan_h.arbitrage(now=LONG_AGO, **ARGS)
|
|
new_v = plan_h.arbitrage(now=now, **ARGS)
|
|
check("without `now` (the old verdict) the buy leg is the ended 03:00 hour "
|
|
"(first of the tie) and decide() does nothing at 04:05 - the pause",
|
|
old_v.cheapest.start == H3
|
|
and arbitrage._act(plan_h, now, old_v, 50.0, 2500.0, 2500.0, 10.0,
|
|
90.0, 5000.0).action is None)
|
|
check("with `now` the buy leg is 04:00, the hour in progress, and decide() "
|
|
"charges at 04:05: the trade runs through",
|
|
new_v.cheapest.start == H3 + timedelta(hours=1)
|
|
and dec(plan_h, now).action == CHARGE)
|
|
|
|
# The day/night tariff, refresh at :20 of the hour: all 15 day hours tie.
|
|
b_local = D0 + timedelta(hours=10, minutes=20) # 12:20 Brussels
|
|
plan_s = static_plan(OPTS, b_local, b_local + timedelta(hours=48), now=b_local)
|
|
now = D0 + timedelta(hours=11, minutes=5) # 13:05 Brussels
|
|
old_v = plan_s.arbitrage(now=LONG_AGO, **ARGS)
|
|
new_v = plan_s.arbitrage(now=now, **ARGS)
|
|
check("the day/night plan built at 12:20 starts at 12:00; used at 13:05 its "
|
|
"sell leg without `now` is the ended 12:00 hour (first of the tied day "
|
|
"hours) and decide() does nothing - the old hourly pause",
|
|
min(p.start for p in plan_s.read.points) == D0 + timedelta(hours=10)
|
|
and old_v.dearest.start == D0 + timedelta(hours=10)
|
|
and arbitrage._act(plan_s, now, old_v, 50.0, 2500.0, 2500.0, 10.0,
|
|
90.0, 5000.0).action is None)
|
|
check("with `now` the sell leg is 13:00, the hour in progress, and decide() "
|
|
"discharges at 13:05: the trade runs through",
|
|
new_v.dearest.start == D0 + timedelta(hours=11)
|
|
and dec(plan_s, now).action == DISCHARGE)
|
|
ok = True
|
|
for m in (0, 5, 20, 35, 50, 59): # every minute class of the hour
|
|
t = D0 + timedelta(hours=12, minutes=m) # 14:mm Brussels
|
|
v = plan_s.arbitrage(now=t, **ARGS)
|
|
ok = ok and v.dearest.start == D0 + timedelta(hours=12) \
|
|
and dec(plan_s, t).action == DISCHARGE
|
|
check("...and through the whole of 14:00-14:59 the target is the 14:00 hour "
|
|
"and the trade stays on (no pause at the top of the hour)", ok)
|
|
|
|
# The exact boundary: a period ends AT its `end`, so at that instant it is over.
|
|
# (`end >= now` would keep it for one more instant.)
|
|
print("\n3c. the exact end of a period")
|
|
at = H3 + timedelta(minutes=15) # 03:15:00 exactly
|
|
v = BUY_PLAN.arbitrage(now=at, **ARGS)
|
|
check("at 03:15:00 exactly the 03:00 quarter is over: the buy leg is 03:15 "
|
|
"and decide() charges", v.cheapest.start == at and dec(BUY_PLAN, at)
|
|
.action == CHARGE)
|
|
check("one microsecond earlier the 03:00 quarter is still in view",
|
|
BUY_PLAN.arbitrage(now=at - timedelta(microseconds=1), **ARGS)
|
|
.cheapest.start == H3)
|
|
at_h = H3 + timedelta(hours=1) # 04:00:00 exactly
|
|
check("at 04:00:00 exactly the 03:00 hour is over (hourly feed), one "
|
|
"microsecond earlier it is not",
|
|
plan_h.arbitrage(now=at_h, **ARGS).cheapest.start == at_h
|
|
and plan_h.arbitrage(now=at_h - timedelta(microseconds=1), **ARGS)
|
|
.cheapest.start == H3)
|
|
two = qplan([0.10, 0.20]) # two firm quarters from 02:00
|
|
check("the Home page's refusal key at the exact end of the first of two "
|
|
"quarters: one is left, so 'not_firm' (with `end >= now` it would say "
|
|
"'not_worth'); a microsecond earlier both count",
|
|
homeview.refusal_key(two, False, D0 + timedelta(minutes=15)) == "not_firm"
|
|
and homeview.refusal_key(two, False, D0 + timedelta(minutes=15)
|
|
- timedelta(microseconds=1)) == "not_worth")
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
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(now=LONG_AGO, **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)
|
|
import inspect # noqa: E402
|
|
|
|
_par = inspect.signature(TariffPlan.arbitrage).parameters
|
|
check("`now` is required: keyword-only with no default, like the three "
|
|
"arguments beside it, so a caller that forgets it fails loudly instead "
|
|
"of getting the elapsed periods back",
|
|
_par["now"].default is inspect.Parameter.empty
|
|
and _par["now"].kind is inspect.Parameter.KEYWORD_ONLY
|
|
and all(p.default is inspect.Parameter.empty
|
|
for n, p in _par.items() if n != "self"))
|
|
try:
|
|
BUY_PLAN.arbitrage(**ARGS_NO_NOW)
|
|
_forgot = False
|
|
except TypeError:
|
|
_forgot = True
|
|
check("...and calling without it raises TypeError", _forgot)
|
|
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("\n6. ties: the first extreme still ahead")
|
|
def acts(plan, instants, ago=False, args=ARGS):
|
|
out = []
|
|
for t in instants:
|
|
v = plan.arbitrage(now=LONG_AGO if ago else t, **args)
|
|
d = arbitrage._act(plan, t, v, 50.0, 2500.0, 2500.0, 10.0, 90.0,
|
|
5000.0) if v.kind == "allowed" else None
|
|
out.append(None if d is None else d.action)
|
|
return out
|
|
|
|
|
|
# Hour-constant 15-minute data: the cheap hour is four equal quarters.
|
|
HC = [round(0.40 - 0.0001 * i, 4) for i in range(192)] # falling, no ties
|
|
for q in range(4, 8):
|
|
HC[q] = 0.05 # 03:00-04:00 four equal quarters
|
|
HC[68:72] = [0.50] * 4 # 19:00-20:00 four equal quarters
|
|
HC_PLAN = qplan(HC)
|
|
probe = [H3 + timedelta(minutes=m) for m in (5, 20, 35, 50, 65)]
|
|
old = acts(HC_PLAN, probe, ago=True)
|
|
new = acts(HC_PLAN, probe)
|
|
check(f"before: four equal cheap quarters, the verdict is the first of them, "
|
|
f"so the trade acts in one quarter ({old})",
|
|
old == [CHARGE, None, None, None, None])
|
|
check(f"after: the first one still ahead, so it acts in all four and stops "
|
|
f"with the hour ({new}) - four acting periods for the same hour, one "
|
|
f"block either way",
|
|
new == [CHARGE, CHARGE, CHARGE, CHARGE, None])
|
|
changes = lambda a: sum(1 for x, y in zip(a, a[1:]) if x != y) # noqa: E731
|
|
check("...the action changes the same number of times (one start, one stop)",
|
|
changes(old) == changes(new) == 1)
|
|
|
|
# A real 15-minute day, as production runs it: refresh every 30 min from the
|
|
# floored hour, the verdict at 7 min into every period.
|
|
REAL = os.path.join(HERE, "scenarios",
|
|
"energy-charts-be-price-2026-09-09_2026-09-10-twoday.json")
|
|
_d = json.load(open(REAL))
|
|
_pts = [PricePoint(brussels(datetime.fromtimestamp(t, UTC)), p / 1000.0, False)
|
|
for t, p in zip(_d["unix_seconds"], _d["price"])]
|
|
_feed = PriceRead.series(_pts, PriceProduct.DAY_AHEAD, "generic_dayahead",
|
|
_pts[0].start)
|
|
_day0 = _pts[0].start.astimezone(UTC)
|
|
|
|
def real_day(ago):
|
|
out = []
|
|
for q in range(96):
|
|
t = _day0 + timedelta(minutes=15 * q + 7)
|
|
ref = t.replace(minute=0 if t.minute < 30 else 30, second=0,
|
|
microsecond=0)
|
|
plan = TariffPlan(fill_from_yesterday(
|
|
_feed, yesterday=PriceRead.no_data("none"), since=ref,
|
|
until=ref + timedelta(hours=36), stamp=ref), None, "provider")
|
|
v = plan.arbitrage(now=LONG_AGO if ago else t, **ARGS5)
|
|
out.append(arbitrage._act(plan, t, v, 50.0, 2500.0, 2500.0, 10.0,
|
|
90.0, 5000.0).action
|
|
if v.kind == "allowed" else None)
|
|
return out
|
|
|
|
|
|
r_old, r_new = real_day(True), real_day(False)
|
|
n_old = sum(1 for a in r_old if a)
|
|
n_new = sum(1 for a in r_new if a)
|
|
print(f" real 15-minute day 2026-09-09: acting periods {n_old} -> {n_new}, "
|
|
f"action changes {changes(r_old)} -> {changes(r_new)}")
|
|
check("a real 15-minute day (energy-charts, 2026-09-09) does not switch more "
|
|
"often: the action changes no more often than before",
|
|
changes(r_new) <= changes(r_old))
|
|
check("...and every period the old verdict acted in is still acted in, with "
|
|
"the quarters it had lost to the elapsed-period tie added",
|
|
all(n is not None for o, n in zip(r_old, r_new) if o) and n_new >= n_old)
|
|
|
|
# Alternating hourly prices (the flapping day): ties everywhere, so the verdict
|
|
# moves on every hour. At most once an hour, never once a tick.
|
|
FL = [0.20 + (0.101 if i % 2 else 0.0) for i in range(24)] * 2
|
|
FL_PLAN = qplan(FL, minutes=60)
|
|
ticks = [acts(FL_PLAN, [D0 + timedelta(seconds=30 * k)], args=ARGS5)[0]
|
|
for k in range(24 * 120)]
|
|
check(f"alternating hourly prices: {changes(ticks)} action changes over "
|
|
f"2880 ticks - at most one an hour, not one a tick",
|
|
changes(ticks) <= 24 and CHARGE in ticks and DISCHARGE in ticks)
|
|
|
|
# The Home page path: `homeview.build_plan()` hands its own `now` to
|
|
# `trade_verdict`. With a `now` floored to the hour it would still aim at the
|
|
# 03:00 period at 03:20 and draw a trade that the control cycle does not take.
|
|
print("\n4b. the Home page plan (build_plan) uses the instant it is rendered at")
|
|
c.o["arbitrage_enabled"] = True
|
|
c.modes_on = False
|
|
|
|
|
|
def home_at(t):
|
|
c.tariff_plan, c.tariff_at = GATE_PLAN, t
|
|
return homeview.build_plan(c, t, False, True)
|
|
|
|
|
|
h_early = home_at(H3 + timedelta(minutes=5))
|
|
h_late = home_at(H3 + timedelta(minutes=20))
|
|
trade_kinds = ("grid", "give")
|
|
check("build_plan at 03:05 (03:00 still current): the spread is real, so the "
|
|
"page draws a trade (verdict_key 'trade')",
|
|
h_early["verdict_key"] == "trade"
|
|
and any(r["kind"] in trade_kinds for r in h_early["runs"]))
|
|
check("build_plan at 03:20 (only flat periods ahead): no trade on the page, "
|
|
"the same refusal as trade_verdict gives (verdict_key 'house')",
|
|
h_late["verdict_key"] == "house"
|
|
and not any(r["kind"] in trade_kinds for r in h_late["runs"])
|
|
and h_late["lines"] != h_early["lines"])
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: " + "; ".join(fails))
|
|
sys.exit(1)
|
|
print("all checks passed")
|