tliu93/home-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 12 Wiki Activity

M6-T03: add pricing profile framework + strategy registry

Browse Source 
- 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).
This commit is contained in:
tliu93
2026-06-23 20:58:34 +02:00
parent df54f5518b
commit bedea196c3
8 changed files with 1695 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/integrations/pricing/strategies.py
+288
View File
@@ -0,0 +1,288 @@
"""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)