Files
goodwe-addon/goodwe_controller/app/custom_formula.py
T
glenn schrooyenandClaude Opus 5.5 63dda4b0d3
checks / unit (pull_request) Successful in 19m51s
infra-pin / fresh (pull_request) Successful in 2s
MAINT-20 part 1 review: restore three lost lines, rewrap ragged comments
Comment, docstring and DECISIONS only (same-code script OK; python -OO
bytecode now identical for all six files).
- peakguard.charge_room_w: one-line docstring + SAFETY-125 pointer (cut whole).
- peakguard.shaping_suspended: keep 'the one place this conjunction is
  written; do not inline it'.
- price_forecast module note: an hour with no price and no yesterday is left
  out, never filled (test_price.py 14d cites the module note for it).
- Refill comment blocks the cut left ragged; DECISIONS pointers on one line.
- FCST-04 note: _yesterday_of(), not _previous_day(); FCST-03 note: the 49
  stronger cases are why the rule is a real change.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
2026-11-01 03:02:51 +01:00

484 lines
21 KiB
Python

"""TARIFF-15 - the customer's own tariff card, as two fixed coefficients.
A Belgian dynamic contract prices energy as `B x index + A` with coefficients
that exist only on the customer's card. The customer types those two numbers;
the add-on does every unit conversion and the VAT multiplication. Fixed
coefficients, no expression evaluation of any kind (the string is settable over
HTTP, and an expression cannot be graded against the card). An unknown key is
refused, never ignored. validate.py refuses to start unless the formula
reproduces the card's printed estimate within CARD_TOLERANCE: that grades the
coefficients against the card, not the card against reality.
The provider declares `{ENERGY, MARKUP}`, never `ALL_IN` (a card's formula is
the energy component only). TARIFF-16 added the injection leg through this same
parser. DECISIONS/TARIFF-15-the-card-is-the-grader.md and
DECISIONS/TARIFF-16-the-injection-leg.md."""
import json
import math
from dataclasses import dataclass
from datetime import date, datetime, timezone
from . import messages
class FormulaInvalid(Exception):
"""`price_formula` cannot be used. Raised by the parser, refused by
validate.py at startup - never at 03:00 on a fetch."""
# The enumerations. Each is a CLOSED set: an unknown value is refused by name,
# not guessed.
# unit -> EUR/kWh per one of it. The whole factor-1000 class lives here: tiny,
# explicit, no default.
UNITS: dict = {
"c_eur_per_kwh": 0.01, # eurocent per kWh - Eneco and ENGIE print this
"eur_per_kwh": 1.0, # some aggregators print this
"eur_per_mwh": 0.001, # the wholesale convention, occasionally reused
}
RESOLUTIONS: tuple = ("hourly", "quarter_hourly")
# index id -> the series it names. A second index is a new row here plus a
# fetch path.
INDEXES: dict = {
"epex_be": "Belgian day-ahead price, EUR/MWh",
}
# 0.01 in the card's own units, the last digit both real cards print (Eneco
# 15,55 vs 15.5471; ENGIE 7,522 vs 7.5218): the card's rounding and nothing
# more. DECISIONS/TARIFF-16-the-injection-leg.md.
CARD_TOLERANCE = 0.01
# The all-in plausibility band, EUR/kWh: the low end, where a factor-1000 error
# falls off.
PLAUSIBLE_LO_EUR_KWH = 0.02
PLAUSIBLE_HI_EUR_KWH = 1.00
# ~6 months. A WARNING, never a refusal: ENGIE freezes a contract's
# coefficients.
CARD_STALE_DAYS = 183
_CARD_KEYS = ("month", "index_eur_per_mwh", "expect_incl_vat")
_TOP_KEYS = ("index", "resolution", "factor", "offset", "units", "vat_pct",
"card")
# TARIFF-16 - the SECOND leg, and the one parameter that carries it
EXPORT_FORMULA_OPTION = "price_export_formula"
# The pass-through fraction, `factor x UNITS[units] / 0.001`: index-free and
# sign-free, so a coefficient typed 100x off moves it by 100 whatever the
# price.
EUR_KWH_PER_EUR_MWH = 0.001
# The two plausibility bounds for an INJECTION leg; neither looks at the sign
# of the price.
PASSTHROUGH_LO = 0.10 # 10% of the wholesale price
PASSTHROUGH_HI = 10.0 # 1000% of it
INJECTION_OFFSET_MAX_EUR_KWH = 0.20
@dataclass(frozen=True)
class _Leg:
"""What ONE option's coefficients price, as sentences its refusals read.
`option` is a parameter, not a fork of the parser (TARIFF-16, as TARIFF-17 did
for `parse_static_profile()`)."""
option: str
what: str # "consumption" / "injection", for a sentence
example: str # a VALID blob for this leg, in the JSON-error hint
when: str # when this option is read at all
flat: str # where a FLAT price of this kind already lives
zero_factor: str # what a factor of 0 actually costs on this leg
neg_factor: str # why a negative FACTOR is wrong on this leg
card_required: bool
card_note: str # why it is required, or what is lost without it
card_causes: str # the leg-specific tail of `card_refusal()`
_LEGS = {
"price_formula": _Leg(
option="price_formula",
what="consumption",
example=('{"index":"epex_be","resolution":"hourly","factor":0.102,'
'"offset":1.00,"units":"c_eur_per_kwh","vat_pct":6,'
'"card":{"month":"2026-09","index_eur_per_mwh":133.9905,'
'"expect_incl_vat":15.55}}'),
when="Leave it empty unless “Dynamic price source” is 'custom_formula'.",
zero_factor=" and trading never finds a price difference",
flat="use “Fixed or day/night tariff” (static_profile) instead",
neg_factor="Put the minus sign on the offset, not on the factor",
card_required=True,
card_note="The add-on needs it to check your two numbers",
card_causes=("the units, the VAT, and that network charges are not "
"in the offset"),
),
EXPORT_FORMULA_OPTION: _Leg(
option=EXPORT_FORMULA_OPTION,
what="injection",
example=('{"index":"epex_be","resolution":"hourly","factor":0.1,'
'"offset":-1.188,"units":"c_eur_per_kwh","vat_pct":0}'),
when="Fill it in only if your injection price follows the market.",
zero_factor="",
flat="use “Injection price” (export_price_eur_kwh) instead",
neg_factor="Put the minus sign on the offset, not on the factor",
card_required=False,
card_note=("It is optional here. Without it the add-on only checks "
"the size of your two numbers"),
card_causes="the units, and that the VAT is 0 for injection",
),
}
@dataclass(frozen=True)
class FormulaSpec:
"""One contract's published arithmetic: SEVEN numbers and no code. Every field is
a fixed named scalar; `parse_price_formula()` refuses an unrecognised key."""
index: str
resolution: str
factor: float
offset: float
units: str
vat_pct: float
card_month: str = "" # "YYYY-MM", as the card is dated
card_index_eur_per_mwh: float = None
card_expect_incl_vat: float = None # in `units`, as the card prints it
# TARIFF-16. Which OPTION this spec was read from, so refusals name the leg
# typed.
option: str = "price_formula"
@property
def unit_to_eur_kwh(self) -> float:
return UNITS[self.units]
@property
def card_graded(self) -> bool:
"""Is there a printed estimate to grade these coefficients against? False is a
real state on the injection leg (Eneco prints none); never False on `price_formula`."""
return self.card_expect_incl_vat is not None
@property
def passthrough(self) -> float:
"""The dimensionless share of the wholesale price this passes through. Index-free
and sign-free, which lets `implausible_injection()` bound a leg that goes negative."""
return self.factor * self.unit_to_eur_kwh / EUR_KWH_PER_EUR_MWH
@property
def offset_eur_kwh(self) -> float:
"""The offset in EUR/kWh, whatever units the card printed it in."""
return self.offset * self.unit_to_eur_kwh
# The arithmetic. Two functions, and the SECOND one is the contract.
def in_card_units(spec: FormulaSpec, index_eur_per_mwh: float) -> float:
"""`(factor x index + offset) x (1 + vat_pct/100)`, in the card's units: the line
the card prints. The only place the coefficients are combined, so the card
check and the price path cannot diverge."""
return ((spec.factor * float(index_eur_per_mwh) + spec.offset)
* (1.0 + spec.vat_pct / 100.0))
def price_eur_kwh(spec: FormulaSpec, index_eur_per_mwh: float) -> float:
"""The energy+markup price for one index value, in EUR/kWh. The unit conversion
happens here, once: those steps are the ones that cost a factor of 10, 100 or 1000."""
return in_card_units(spec, index_eur_per_mwh) * spec.unit_to_eur_kwh
def card_discrepancy(spec: FormulaSpec) -> float:
"""How far the coefficients land from the card's printed estimate, in the card's
units, signed: positive means MORE than the card prints (the direction an
`offset` carrying network charges moves it). Raises on a spec with no card
rather than returning 0.0 (TARIFF-16)."""
if not spec.card_graded:
raise FormulaInvalid(
f"{spec.option} has no card block, so there is no printed "
f"estimate to measure a discrepancy against. Ask "
f"`spec.card_graded` before calling this.")
return in_card_units(spec, spec.card_index_eur_per_mwh) \
- spec.card_expect_incl_vat
# Reading the option
def _finite(value, field: str, option: str = "price_formula") -> float:
"""A real, finite number, or a refusal naming the field (NaN would price every hour NaN; `True` is an `int`)."""
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise FormulaInvalid(
f"{option}.{field} is {value!r}, which is not a number. Write a "
f"plain number, for example 0.102.")
if not math.isfinite(float(value)):
raise FormulaInvalid(
f"{option}.{field} is {value!r}, which is not finite. Write a "
f"plain number, for example 0.102.")
return float(value)
def _card(raw, leg: _Leg = None) -> tuple:
"""The `card` block: the estimate the coefficients are graded against. Required on
the consumption leg, optional on the injection one because Eneco prints none
(`("", None, None)`; `FormulaSpec.card_graded`)."""
if leg is None:
leg = _LEGS["price_formula"]
if raw is None:
if not leg.card_required:
return ("", None, None)
raise FormulaInvalid(
f"{leg.option} has no \"card\" block. {leg.card_note}. Copy "
"the three figures from your tariff card, for example "
"{\"month\": \"2026-09\", \"index_eur_per_mwh\": 133.9905, "
"\"expect_incl_vat\": 15.55}.")
if not isinstance(raw, dict):
raise FormulaInvalid(
f"{leg.option}.card is {type(raw).__name__}, not an object with "
f"{', '.join(_CARD_KEYS)}.")
unknown = sorted(set(raw) - set(_CARD_KEYS))
if unknown:
raise FormulaInvalid(
f"{leg.option}.card has unknown field(s) {unknown}. It takes "
f"only {list(_CARD_KEYS)}.")
month = str(raw.get("month", "") or "").strip()
if not _month_date(month):
raise FormulaInvalid(
f"{leg.option}.card.month is {raw.get('month')!r}. Write the "
f"month of the card as \"YYYY-MM\", for example \"2026-09\".")
return (month,
_finite(raw.get("index_eur_per_mwh"), "card.index_eur_per_mwh",
leg.option),
_finite(raw.get("expect_incl_vat"), "card.expect_incl_vat",
leg.option))
def _month_date(month: str):
"""`"2026-09"` -> the first of that month, or None if it is not one."""
try:
year, mon = str(month).split("-")
return date(int(year), int(mon), 1)
except (TypeError, ValueError):
return None
def parse_price_formula(value, *, option: str = "price_formula"):
"""`price_formula` -> a `FormulaSpec`, or None when unset. `option` selects a row of
`_LEGS`; everything else is shared, so `parse_export_formula()` is a wrapper."""
leg = _LEGS[option]
if value is None:
return None
if isinstance(value, str):
text = value.strip()
if not text or text.lower() in ("null", "none", "off"):
return None
try:
value = json.loads(text)
except ValueError as e:
raise FormulaInvalid(
f"{messages.not_valid(option, e)}. Copy the example from the "
f"documentation. {leg.when}")
if not isinstance(value, dict):
raise FormulaInvalid(
f"{option} is {type(value).__name__}, not a JSON object with "
f"{', '.join(_TOP_KEYS)}.")
# An unknown key is refused, not ignored: a future shape is a NAMED field,
# never a grammar.
unknown = sorted(set(value) - set(_TOP_KEYS))
if unknown:
raise FormulaInvalid(
f"{option} has unknown field(s) {unknown}. It takes only "
f"{list(_TOP_KEYS)}.")
index = str(value.get("index", "") or "").strip()
if index not in INDEXES:
raise FormulaInvalid(
f"{option}.index is {value.get('index')!r}. The add-on can only "
f"fetch "
f"{', '.join(f'{k!r} ({v})' for k, v in sorted(INDEXES.items()))}.")
resolution = str(value.get("resolution", "") or "").strip()
if resolution not in RESOLUTIONS:
raise FormulaInvalid(
f"{option}.resolution is {value.get('resolution')!r}. Choose one "
f"of {list(RESOLUTIONS)}, as your tariff card says.")
# `units` is required with no default: a default is the factor-1000 class
# with a friendly face.
units = str(value.get("units", "") or "").strip()
if units not in UNITS:
raise FormulaInvalid(
f"{option}.units is {value.get('units')!r}. Choose one of "
f"{sorted(UNITS)}, as printed beside the formula on your card.")
factor = _finite(value.get("factor"), "factor", option)
offset = _finite(value.get("offset"), "offset", option)
vat_pct = _finite(value.get("vat_pct"), "vat_pct", option)
# Zero is a fixed tariff in a dynamic one's clothes (arbitrage never
# fires); point the operator at the first-class fixed price.
if factor == 0.0:
raise FormulaInvalid(
f"{option}.factor is 0, so every hour gets the same {leg.what} "
f"price{leg.zero_factor}. For one flat price, {leg.flat}.")
# A negative factor inverts the market: charge into the evening peak,
# discharge at 03:00.
if factor < 0.0:
raise FormulaInvalid(
f"{option}.factor is {factor:g}, which turns the market upside "
f"down. {leg.neg_factor}.")
# 0 is legitimate (injection is VAT-exempt; some cards are VAT-inclusive);
# the bound is for the 106 / 1.06 / 6 confusion.
if not 0.0 <= vat_pct <= 30.0:
raise FormulaInvalid(
f"{option}.vat_pct is {vat_pct:g}. Write a percentage between 0 "
f"and 30, for example 6.")
month, card_index, card_expect = _card(value.get("card"), leg)
spec = FormulaSpec(index=index, resolution=resolution, factor=factor,
offset=offset, units=units, vat_pct=vat_pct,
card_month=month, card_index_eur_per_mwh=card_index,
card_expect_incl_vat=card_expect, option=option)
# The card self-check still runs wherever there is a card, with the same
# tolerance. Where the card prints no estimate, `implausible_injection()`
# bounds the coefficients and a startup line says the leg is ungraded.
if spec.card_graded:
delta = card_discrepancy(spec)
if abs(delta) > CARD_TOLERANCE:
raise FormulaInvalid(card_refusal(spec, delta))
if option == EXPORT_FORMULA_OPTION:
band = implausible_injection(spec)
if band:
raise FormulaInvalid(band)
return spec
def parse_export_formula(value):
"""`price_export_formula` -> the INJECTION leg's `FormulaSpec`, or None (TARIFF-16).
Same parser with `option` moved; the flat `export_price_eur_kwh()` still answers
when unset. Nothing trades on it today. DECISIONS/TARIFF-16-the-injection-leg.md."""
return parse_price_formula(value, option=EXPORT_FORMULA_OPTION)
def card_refusal(spec: FormulaSpec, delta: float) -> str:
"""The sentence an operator reads when their formula misses its own card: both
numbers and what to check (causes are in DOCS.md)."""
got = in_card_units(spec, spec.card_index_eur_per_mwh)
leg = _LEGS[spec.option]
return (
f"{spec.option} gives {got:.4f} "
f"{messages.UNIT_WORDS.get(spec.units, spec.units)}, the card says "
f"{spec.card_expect_incl_vat:g}. Check {leg.card_causes}.")
# The two questions the PLUGIN cannot ask, because they need framework numbers
def implausible_all_in(spec: FormulaSpec, taxes_eur_kwh: float):
"""The card's own point, if its LEVEL cannot be a Belgian retail price. Refusal
sentence or None. Called from validate.py only (it reads the framework-owned
taxes). The low end is graded on the energy component and the high end on the
all-in figure; the consumption leg's band only. It grades the card's own point,
not the fetched series. DECISIONS/TARIFF-16-the-injection-leg.md."""
energy = price_eur_kwh(spec, spec.card_index_eur_per_mwh)
all_in = energy + float(taxes_eur_kwh)
if energy >= PLAUSIBLE_LO_EUR_KWH and all_in <= PLAUSIBLE_HI_EUR_KWH:
return None
return (
f"{spec.option} gives {energy:.5f} EUR/kWh for energy and "
f"{all_in:.5f} EUR/kWh with “Distribution, levies and VAT” "
f"(price_taxes_eur_kwh), outside the plausible band of "
f"{PLAUSIBLE_LO_EUR_KWH}-{PLAUSIBLE_HI_EUR_KWH} EUR/kWh. Check \"units\" against the unit beside your formula.")
def implausible_injection(spec: FormulaSpec):
"""The INJECTION leg's plausibility band. Refusal or None. A negative price is
accepted; the bounds are on the coefficients (pass-through and offset
magnitude), since a band spanning zero cannot catch a card 100x too small. It
bounds the shape, not the truth. DECISIONS/TARIFF-16-the-injection-leg.md."""
if not PASSTHROUGH_LO <= spec.passthrough <= PASSTHROUGH_HI:
return (
f"{spec.option}.factor passes through {spec.passthrough:.4g}x the "
f"market price, outside {PASSTHROUGH_LO}-{PASSTHROUGH_HI}x. Check "
f"\"units\" ({spec.units!r}) against the unit beside your "
f"formula.")
if abs(spec.offset_eur_kwh) > INJECTION_OFFSET_MAX_EUR_KWH:
return (
f"{spec.option}.offset is {spec.offset:g} {spec.units}, which is "
f"{spec.offset_eur_kwh:+.5f} EUR/kWh and larger than "
f"{INJECTION_OFFSET_MAX_EUR_KWH} EUR/kWh either way. Check its "
f"unit.")
return None
def stale_card(spec: FormulaSpec, today=None):
"""A warning when the card is older than ~6 months, or None. Never a refusal (ENGIE freezes coefficients)."""
dated = _month_date(spec.card_month)
now = today or datetime.now(timezone.utc).date()
if dated is None or (now - dated).days <= CARD_STALE_DAYS:
return None
return (
f"{spec.option}'s card is dated {spec.card_month}, "
f"{(now - dated).days} days ago. The add-on still starts, but if your "
f"contract changed since, copy the new figures from your supplier.")
# Resolution
def to_resolution(pairs, resolution: str) -> list:
"""`[(instant, index EUR/MWh)]` -> the series the CONTRACT is written on.
`quarter_hourly` returns the feed's points, `hourly` averages each UTC hour
(exact for Belgium: whole-hour offsets) at the top of the hour. An incomplete
hour is averaged over what is there. It sets how many points are emitted, not
how long one covers (TARIFF-14)."""
ordered = sorted(pairs, key=lambda r: r[0].astimezone(timezone.utc))
if resolution == "quarter_hourly":
return list(ordered)
if resolution != "hourly":
raise FormulaInvalid(f"unknown resolution {resolution!r}")
buckets: dict = {}
for start, value in ordered:
key = start.astimezone(timezone.utc).replace(
minute=0, second=0, microsecond=0)
buckets.setdefault(key, []).append(value)
return [(k, sum(v) / len(v)) for k, v in sorted(buckets.items())]
# The startup preview
PREVIEW_HOURS = 24
PREVIEW_MAX_ROWS = 24
def preview_lines(spec: FormulaSpec, rows) -> list:
"""`[(instant, index EUR/MWh, all-in EUR/kWh)]` -> lines for the log: the index
that went IN beside the price that came OUT. Prints a bounded row COUNT, not
every fourth point (that would assume the granularity)."""
if not rows:
return []
shown = rows[:PREVIEW_MAX_ROWS]
prices = sorted(r[2] for r in rows)
mid = len(prices) // 2
median = (prices[mid] if len(prices) % 2
else (prices[mid - 1] + prices[mid]) / 2.0)
out = [
f"custom_formula preview - the next {PREVIEW_HOURS} h of YOUR tariff, "
f"index EUR/MWh -> all-in EUR/kWh, computed as "
f"({spec.factor:g} x index + {spec.offset:g}) x "
f"(1 + {spec.vat_pct:g}/100) {spec.units}, converted to EUR/kWh and "
f"with price_taxes_eur_kwh added by the framework:"]
for start, index, all_in in shown:
out.append(f" {start.isoformat()} {index:9.3f} -> {all_in:9.5f}")
if len(rows) > len(shown):
out.append(f" ... {len(rows) - len(shown)} further point(s) not "
f"listed; the figures below cover ALL {len(rows)}")
out.append(
f" min {prices[0]:.5f} / median {median:.5f} / max {prices[-1]:.5f} "
f"EUR/kWh, spread {prices[-1] - prices[0]:.5f} EUR/kWh over "
f"{len(rows)} point(s). ⚠️ If those are a thousand times what your "
f"bill says, `units` is wrong; if the spread is 0, `factor` is.")
return out