home-automation/app/integrations/pricing/strategies.py

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