"""Price-calculation strategy registry and built-in implementations. A *strategy* is a plain function registered under a ``kind`` string. Given per-register kWh deltas, the period start time, the active contract-version values, and a DB session, it returns a result dict with: { "import_cost": Decimal, # total cost of electricity drawn from grid "export_revenue": Decimal, # total revenue from electricity fed to grid "net_cost": Decimal, # import_cost − export_revenue "pricing": dict, # snapshot of the price inputs used (for auditing) } **Import and export are kept separate throughout** — net_cost is only derived at the end. This supports the Dutch no-netting rule (saldering afgebouwd). **All monetary arithmetic uses Decimal** converted via ``Decimal(str(x))`` to avoid float binary rounding errors. Design notes ------------ - **Purely data + functions**: no ABC, no class hierarchy. A ``dict``-based registry is the simplest structure that supports the two current kinds and leaves the door open for future additions (e.g. ``octopus``, ``frank``). - ``deltas`` is a plain dataclass ``PeriodDeltas`` — typed, but no ORM. - Tibber strategy queries ``TibberPrice`` directly from the session; it does not call the Tibber API (that is T05's job). """ from __future__ import annotations import logging from dataclasses import dataclass from datetime import datetime from decimal import Decimal from typing import Any, Callable from sqlalchemy.orm import Session logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Shared data types # --------------------------------------------------------------------------- @dataclass class PeriodDeltas: """Per-register kWh deltas for one 15-minute billing period. All values are ``Decimal`` and represent ``end_reading − start_reading`` for the corresponding cumulative energy register. Attributes ---------- d1: Decimal Delivered (consumed) kWh on the low/dal tariff register (delivered_1). d2: Decimal Delivered (consumed) kWh on the normal/high tariff register (delivered_2). r1: Decimal Returned (fed-to-grid) kWh on the low/dal tariff register (returned_1). r2: Decimal Returned (fed-to-grid) kWh on the normal/high tariff register (returned_2). """ d1: Decimal # delivered low-tariff d2: Decimal # delivered high-tariff r1: Decimal # returned low-tariff r2: Decimal # returned high-tariff # Strategy callable signature. StrategyFn = Callable[ [PeriodDeltas, datetime, dict[str, Any], Session], dict[str, Any], ] # --------------------------------------------------------------------------- # Registry # --------------------------------------------------------------------------- _REGISTRY: dict[str, StrategyFn] = {} def register_strategy(kind: str, fn: StrategyFn) -> None: """Register *fn* as the price-calculation strategy for *kind*. Overwrites any previously registered strategy for the same kind (allows monkey-patching in tests). """ _REGISTRY[kind] = fn def get_strategy(kind: str) -> StrategyFn: """Return the registered strategy for *kind*. Raises ------ KeyError If no strategy is registered for *kind*. """ if kind not in _REGISTRY: raise KeyError( f"No price strategy registered for kind={kind!r}. " f"Available: {sorted(_REGISTRY)}" ) return _REGISTRY[kind] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _to_decimal(value: Any) -> Decimal: """Convert *value* to Decimal via str() to avoid float binary rounding.""" return Decimal(str(value)) # --------------------------------------------------------------------------- # Manual strategy — fixed dual-tariff # --------------------------------------------------------------------------- def _manual_strategy( deltas: PeriodDeltas, t0: datetime, values: dict[str, Any], session: Session, ) -> dict[str, Any]: """Calculate billing costs for one period using fixed dual-tariff rates. Formula (§3.4): - ``buy_dal = energy.buy.dal + energy.energy_tax + energy.ode`` - ``buy_normal = energy.buy.normal + energy.energy_tax + energy.ode`` - ``import_cost = Δd1 × buy_dal + Δd2 × buy_normal`` - ``sell_dal = energy.sell.dal`` (no tax on return price) - ``sell_normal = energy.sell.normal`` - ``export_revenue = Δr1 × sell_dal + Δr2 × sell_normal`` - ``net_cost = import_cost − export_revenue`` The ``pricing`` snapshot contains all per-unit prices used so that the result is fully auditable without re-querying the contract version. """ energy = values.get("energy", {}) buy = energy.get("buy", {}) sell = energy.get("sell", {}) energy_tax = _to_decimal(energy.get("energy_tax", 0)) ode = _to_decimal(energy.get("ode", 0)) buy_dal_base = _to_decimal(buy.get("dal", 0)) buy_normal_base = _to_decimal(buy.get("normal", 0)) sell_dal = _to_decimal(sell.get("dal", 0)) sell_normal = _to_decimal(sell.get("normal", 0)) # Effective buy prices including all taxes. buy_dal = buy_dal_base + energy_tax + ode buy_normal = buy_normal_base + energy_tax + ode import_cost = deltas.d1 * buy_dal + deltas.d2 * buy_normal export_revenue = deltas.r1 * sell_dal + deltas.r2 * sell_normal net_cost = import_cost - export_revenue pricing_snapshot = { "kind": "manual", "buy_dal": str(buy_dal), "buy_normal": str(buy_normal), "sell_dal": str(sell_dal), "sell_normal": str(sell_normal), "energy_tax": str(energy_tax), "ode": str(ode), } return { "import_cost": import_cost, "export_revenue": export_revenue, "net_cost": net_cost, "pricing": pricing_snapshot, } # --------------------------------------------------------------------------- # Tibber strategy — dynamic 15-minute spot price # --------------------------------------------------------------------------- class TibberPriceNotFoundError(LookupError): """Raised when no TibberPrice row covers the requested period start. The billing engine (T07) catches this to mark the period as ``degraded`` or skip it until a price becomes available. """ def _tibber_strategy( deltas: PeriodDeltas, t0: datetime, values: dict[str, Any], session: Session, ) -> dict[str, Any]: """Calculate billing costs for one period using Tibber dynamic pricing. Price lookup: Takes the most recent ``TibberPrice`` row where ``starts_at ≤ t0`` (i.e. the slot that was in effect at *t0*). This is a DESC LIMIT 1 query on ``starts_at``. Formula (§3.4): - ``buy = total`` (Tibber's all-inclusive price; already includes energy tax, VAT and the buy-side ``inkoopvergoeding``) - ``sell = total − energy_tax − sell_fee − sell_adjust`` - ``import_cost = (Δd1 + Δd2) × buy`` - ``export_revenue = (Δr1 + Δr2) × sell`` - ``net_cost = import_cost − export_revenue`` ``sell_fee`` models Tibber's per-kWh **verkoopvergoeding** (feed-in fee, €0.0248/kWh incl. VAT since 2026-01-01). It is always deducted from the feed-in payout: even under the net-metering (saldering) scheme, Tibber pays ``total − verkoopvergoeding`` per returned kWh (Tibber NL: "€0,28 − €0,0248 = €0,2552"). ``total`` already contains the equal buy-side ``inkoopvergoeding``, so the two fees do **not** cancel — the feed-in price sits ``sell_fee`` below the buy price. ``sell_adjust`` is a separate manual correction: under net metering it carries back the refunded energy tax (``sell_adjust = −energy_tax``), leaving ``sell = total − sell_fee``. Tibber does not differentiate tariff slots (dal vs normal) — the 15-minute API price applies to the full delivered/returned volume. Negative ``total`` (extreme negative spot prices): When ``total`` is negative the ``sell`` price will also be negative, meaning ``export_revenue`` becomes negative (feeding to grid *costs* money). This is the mathematically correct outcome and is left as-is. Raises ------ TibberPriceNotFoundError If no ``TibberPrice`` row exists with ``starts_at ≤ t0``. The caller (T07 billing engine) should catch this and mark the period as ``degraded`` or skip it for later recomputation. """ from app.models.energy import TibberPrice # local import to avoid circular from sqlalchemy import desc price_row: TibberPrice | None = ( session.query(TibberPrice) .filter(TibberPrice.starts_at <= t0) .order_by(desc(TibberPrice.starts_at)) .first() ) if price_row is None: raise TibberPriceNotFoundError( f"No TibberPrice found covering t0={t0.isoformat()!r}; " "period cannot be billed until price data is available." ) energy = values.get("energy", {}) energy_tax = _to_decimal(energy.get("energy_tax", 0)) sell_fee = _to_decimal(energy.get("sell_fee", 0)) sell_adjust = _to_decimal(energy.get("sell_adjust", 0)) total = _to_decimal(price_row.total) buy = total sell = total - energy_tax - sell_fee - sell_adjust total_delivered = deltas.d1 + deltas.d2 total_returned = deltas.r1 + deltas.r2 import_cost = total_delivered * buy export_revenue = total_returned * sell net_cost = import_cost - export_revenue pricing_snapshot = { "kind": "tibber", "tibber_price_starts_at": price_row.starts_at.isoformat(), "tibber_price_id": price_row.id, "total": str(total), "buy": str(buy), "sell": str(sell), "energy_tax": str(energy_tax), "sell_fee": str(sell_fee), "sell_adjust": str(sell_adjust), } return { "import_cost": import_cost, "export_revenue": export_revenue, "net_cost": net_cost, "pricing": pricing_snapshot, } # --------------------------------------------------------------------------- # Register built-in strategies at module import time # --------------------------------------------------------------------------- register_strategy("manual", _manual_strategy) register_strategy("tibber", _tibber_strategy)