tliu93
bedea196c3
- app/integrations/pricing: profiles.py (pydantic ManualProfile/TibberProfile, load_profile/list_profiles/validate_values), manual.yaml + tibber.yaml (structure-only, no price values), strategies.py (register/get_strategy, manual dual-tariff + tibber strategies, all Decimal). - tibber strategy matches nearest tibber_price with starts_at <= t0. - tests for profiles + strategies (hand-checked dual-tariff & tibber math).
289 lines
9.6 KiB
Python
289 lines
9.6 KiB
Python
"""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 tax)
|
||
- ``sell = total − energy_tax − sell_adjust``
|
||
- ``import_cost = (Δd1 + Δd2) × buy``
|
||
- ``export_revenue = (Δr1 + Δr2) × sell``
|
||
- ``net_cost = import_cost − export_revenue``
|
||
|
||
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_adjust = _to_decimal(energy.get("sell_adjust", 0))
|
||
|
||
total = _to_decimal(price_row.total)
|
||
buy = total
|
||
sell = total - energy_tax - 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_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)
|