"""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