From bedea196c3ebca60738e665887343e76cf2ffd97 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Tue, 23 Jun 2026 20:58:34 +0200 Subject: [PATCH] M6-T03: add pricing profile framework + strategy registry - 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). --- app/integrations/pricing/__init__.py | 11 + app/integrations/pricing/profiles.py | 393 +++++++++++++ app/integrations/pricing/profiles/manual.yaml | 20 + app/integrations/pricing/profiles/tibber.yaml | 14 + app/integrations/pricing/strategies.py | 288 +++++++++ docs/design/m6-tibber-dynamic-energy.md | 2 +- tests/test_pricing_profiles.py | 415 +++++++++++++ tests/test_pricing_strategies.py | 553 ++++++++++++++++++ 8 files changed, 1695 insertions(+), 1 deletion(-) create mode 100644 app/integrations/pricing/__init__.py create mode 100644 app/integrations/pricing/profiles.py create mode 100644 app/integrations/pricing/profiles/manual.yaml create mode 100644 app/integrations/pricing/profiles/tibber.yaml create mode 100644 app/integrations/pricing/strategies.py create mode 100644 tests/test_pricing_profiles.py create mode 100644 tests/test_pricing_strategies.py diff --git a/app/integrations/pricing/__init__.py b/app/integrations/pricing/__init__.py new file mode 100644 index 0000000..9d6002f --- /dev/null +++ b/app/integrations/pricing/__init__.py @@ -0,0 +1,11 @@ +"""Pricing profile framework and strategy registry. + +This package provides: + +- ``profiles``: Pydantic models describing the structure of pricing profiles + (``ManualProfile`` / ``TibberProfile``), YAML loaders, and contract-values + validation (``load_profile`` / ``list_profiles`` / ``validate_values``). +- ``strategies``: A lightweight registry of price-calculation strategies + (``register_strategy`` / ``get_strategy``), with built-in implementations for + the ``manual`` (fixed dual-tariff) and ``tibber`` (dynamic API) kinds. +""" diff --git a/app/integrations/pricing/profiles.py b/app/integrations/pricing/profiles.py new file mode 100644 index 0000000..42d39d8 --- /dev/null +++ b/app/integrations/pricing/profiles.py @@ -0,0 +1,393 @@ +"""Pricing profile loader, validator, and contract-values checker. + +A *pricing profile* is a YAML file that describes the **structure** of an energy +contract — which fields exist, their units, and which have defaults. Actual +pricing values (the numbers the user fills in via the UI) are stored in the DB +as ``EnergyContractVersion.values`` (a JSON blob) and must conform to the +corresponding profile's structure. + +Profiles live in ``app/integrations/pricing/profiles/.yaml`` and are +located at runtime relative to *this file* (not CWD), matching the pattern +established by the Modbus profile loader. + +Design notes +------------ +- **Purely data + functions** — no abstract base classes or inheritance. +- Two Pydantic models — ``ManualProfile`` and ``TibberProfile`` — capture the + different structures of the two supported kinds. A thin union dispatcher in + ``load_profile`` picks the right one. +- ``validate_values(kind, values)`` fills in fields that carry a ``default`` and + raises ``ProfileValidationError`` for missing required fields or wrong types. +- All errors are subclasses of built-ins so callers need not import this module + just to catch them. +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any, Optional + +import yaml +from pydantic import BaseModel, ValidationError, model_validator + +logger = logging.getLogger(__name__) + +# Directory containing the YAML profiles, located relative to *this* file. +_PROFILES_DIR = Path(__file__).parent / "profiles" + + +# --------------------------------------------------------------------------- +# Custom exceptions +# --------------------------------------------------------------------------- + + +class ProfileNotFoundError(FileNotFoundError): + """Raised when the requested profile YAML file does not exist.""" + + +class ProfileValidationError(ValueError): + """Raised when a profile YAML fails Pydantic validation or when contract + values do not conform to the profile structure.""" + + +# --------------------------------------------------------------------------- +# Leaf-node models (a single pricing field: unit + optional default) +# --------------------------------------------------------------------------- + + +class FieldSpec(BaseModel): + """Specification for a single numeric pricing field.""" + + unit: str + """Physical / monetary unit string (e.g. ``"EUR/kWh"``, ``"EUR/month"``).""" + + default: Optional[float] = None + """Default value used when the field is absent from contract values. + ``None`` means the field is required (no default).""" + + +# --------------------------------------------------------------------------- +# ManualProfile — fixed / variable dual-tariff contract structure +# --------------------------------------------------------------------------- + + +class ManualBuySpec(BaseModel): + normal: FieldSpec # high-tariff buy price (delivered_2 registers) + dal: FieldSpec # low-tariff buy price (delivered_1 registers) + + +class ManualSellSpec(BaseModel): + normal: FieldSpec # high-tariff sell / return price + dal: FieldSpec # low-tariff sell / return price + + +class ManualEnergySpec(BaseModel): + dual_tariff: bool # always True for manual profiles + buy: ManualBuySpec + sell: ManualSellSpec + energy_tax: FieldSpec # added to buy price; includes VAT + ode: FieldSpec # currently merged into energy_tax; default 0 + + +class ManualStandingSpec(BaseModel): + network_fee: FieldSpec # EUR/month + management_fee: FieldSpec # EUR/month + + +class ManualCreditsSpec(BaseModel): + heffingskorting: FieldSpec # EUR/year — energy-tax credit deducted at summary + + +class ManualProfile(BaseModel): + """Complete structure description for a ``manual`` pricing contract.""" + + kind: str + label: str + energy: ManualEnergySpec + standing: ManualStandingSpec + credits: ManualCreditsSpec + + @model_validator(mode="after") + def _check_kind(self) -> "ManualProfile": + if self.kind != "manual": + raise ValueError(f"ManualProfile requires kind='manual', got {self.kind!r}") + return self + + +# --------------------------------------------------------------------------- +# TibberProfile — dynamic Tibber API contract structure +# --------------------------------------------------------------------------- + + +class TibberEnergySpec(BaseModel): + source: str # must be "tibber_api" + energy_tax: FieldSpec # subtracted from total to derive sell price + sell_adjust: FieldSpec # additional sell-price adjustment; default 0 + + +class TibberStandingSpec(BaseModel): + management_fee: FieldSpec # EUR/month; has a default + network_fee: FieldSpec # EUR/month + + +class TibberCreditsSpec(BaseModel): + heffingskorting: FieldSpec # EUR/year + + +class TibberProfile(BaseModel): + """Complete structure description for a ``tibber`` pricing contract.""" + + kind: str + label: str + energy: TibberEnergySpec + standing: TibberStandingSpec + credits: TibberCreditsSpec + + @model_validator(mode="after") + def _check_kind(self) -> "TibberProfile": + if self.kind != "tibber": + raise ValueError(f"TibberProfile requires kind='tibber', got {self.kind!r}") + return self + + +# A union type for type hints where either profile is acceptable. +AnyProfile = ManualProfile | TibberProfile + +# Map kind → Pydantic model class used for validation. +_PROFILE_MODELS: dict[str, type[ManualProfile] | type[TibberProfile]] = { + "manual": ManualProfile, + "tibber": TibberProfile, +} + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def load_profile(kind: str) -> AnyProfile: + """Load and validate a pricing profile by kind name. + + The profile file is expected at + ``app/integrations/pricing/profiles/.yaml``. + + Parameters + ---------- + kind: + Profile kind without extension (``"manual"`` or ``"tibber"``). + + Returns + ------- + ManualProfile | TibberProfile + Validated profile model. + + Raises + ------ + ProfileNotFoundError + If ``profiles/.yaml`` does not exist. + ProfileValidationError + If the YAML is syntactically valid but fails schema validation. + """ + path = _PROFILES_DIR / f"{kind}.yaml" + if not path.exists(): + raise ProfileNotFoundError( + f"Pricing profile '{kind}' not found (looked for {path})" + ) + + with path.open("r", encoding="utf-8") as fh: + raw = yaml.safe_load(fh) + + if not isinstance(raw, dict): + raise ProfileValidationError( + f"Profile '{kind}': expected a YAML mapping, got {type(raw).__name__}" + ) + + # Choose the right Pydantic model based on the ``kind`` field in the YAML. + yaml_kind = raw.get("kind", kind) + model_cls = _PROFILE_MODELS.get(yaml_kind) + if model_cls is None: + raise ProfileValidationError( + f"Profile '{kind}' has unknown kind={yaml_kind!r}; " + f"supported: {list(_PROFILE_MODELS)}" + ) + + try: + return model_cls.model_validate(raw) + except ValidationError as exc: + raise ProfileValidationError( + f"Profile '{kind}' failed validation: {exc}" + ) from exc + + +def list_profiles() -> list[dict[str, Any]]: + """Return structural data for all available pricing profiles. + + Scans ``app/integrations/pricing/profiles/*.yaml``, loads each, and returns + a list of dicts (Pydantic model dumps). Profiles that fail to load are + skipped with a warning. + + Returns + ------- + list[dict[str, Any]] + One entry per valid profile, suitable for JSON serialisation and + front-end form rendering. + """ + results: list[dict[str, Any]] = [] + if not _PROFILES_DIR.exists(): + return results + + for path in sorted(_PROFILES_DIR.glob("*.yaml")): + kind = path.stem + try: + profile = load_profile(kind) + results.append(profile.model_dump()) + except (ProfileNotFoundError, ProfileValidationError, Exception) as exc: + logger.warning("Skipping malformed pricing profile '%s': %s", kind, exc) + + return results + + +# --------------------------------------------------------------------------- +# Contract-values validation +# --------------------------------------------------------------------------- + + +def _fill_defaults_manual(values: dict[str, Any], profile: ManualProfile) -> dict[str, Any]: + """Return a copy of *values* with any ``ode`` default applied if missing.""" + filled = dict(values) + + energy = dict(filled.get("energy", {})) + + # Apply default for ode (default=0) if absent. + if "ode" not in energy and profile.energy.ode.default is not None: + energy["ode"] = profile.energy.ode.default + + filled["energy"] = energy + return filled + + +def _fill_defaults_tibber(values: dict[str, Any], profile: TibberProfile) -> dict[str, Any]: + """Return a copy of *values* with sell_adjust and management_fee defaults applied.""" + filled = dict(values) + + energy = dict(filled.get("energy", {})) + # Apply default for sell_adjust (default=0) if absent. + if "sell_adjust" not in energy and profile.energy.sell_adjust.default is not None: + energy["sell_adjust"] = profile.energy.sell_adjust.default + filled["energy"] = energy + + standing = dict(filled.get("standing", {})) + # Apply default for management_fee if absent. + if ( + "management_fee" not in standing + and profile.standing.management_fee.default is not None + ): + standing["management_fee"] = profile.standing.management_fee.default + filled["standing"] = standing + + return filled + + +def _require_numeric(section: str, key: str, container: dict[str, Any]) -> None: + """Assert that *container[key]* exists and is a number; raise ProfileValidationError.""" + if key not in container: + raise ProfileValidationError( + f"Contract values missing required field '{section}.{key}'" + ) + val = container[key] + if not isinstance(val, (int, float)): + raise ProfileValidationError( + f"Contract values field '{section}.{key}' must be a number, " + f"got {type(val).__name__!r}" + ) + + +def _validate_manual_values(values: dict[str, Any], profile: ManualProfile) -> dict[str, Any]: + """Validate and fill-defaults for manual contract values. + + Returns the filled values dict on success. Raises ProfileValidationError + on missing required fields or wrong types. + """ + filled = _fill_defaults_manual(values, profile) + energy = filled.get("energy", {}) + buy = energy.get("buy", {}) + sell = energy.get("sell", {}) + standing = filled.get("standing", {}) + credits = filled.get("credits", {}) + + # Required energy.buy fields. + _require_numeric("energy.buy", "normal", buy) + _require_numeric("energy.buy", "dal", buy) + # Required energy.sell fields. + _require_numeric("energy.sell", "normal", sell) + _require_numeric("energy.sell", "dal", sell) + # Required energy fields. + _require_numeric("energy", "energy_tax", energy) + _require_numeric("energy", "ode", energy) + # Required standing fields. + _require_numeric("standing", "network_fee", standing) + _require_numeric("standing", "management_fee", standing) + # Required credits fields. + _require_numeric("credits", "heffingskorting", credits) + + return filled + + +def _validate_tibber_values(values: dict[str, Any], profile: TibberProfile) -> dict[str, Any]: + """Validate and fill-defaults for tibber contract values. + + Returns the filled values dict on success. Raises ProfileValidationError + on missing required fields or wrong types. + """ + filled = _fill_defaults_tibber(values, profile) + energy = filled.get("energy", {}) + standing = filled.get("standing", {}) + credits = filled.get("credits", {}) + + # Required energy fields. + _require_numeric("energy", "energy_tax", energy) + _require_numeric("energy", "sell_adjust", energy) + # Required standing fields. + _require_numeric("standing", "management_fee", standing) + _require_numeric("standing", "network_fee", standing) + # Required credits fields. + _require_numeric("credits", "heffingskorting", credits) + + return filled + + +def validate_values(kind: str, values: dict[str, Any]) -> dict[str, Any]: + """Validate a contract-values dict against the named profile structure. + + Fields that carry a ``default`` in the profile (e.g. ``ode``, + ``sell_adjust``, tibber ``management_fee``) are silently filled in when + absent from *values*. Fields with no default that are absent, or fields + whose value is not a number, cause a ``ProfileValidationError``. + + Parameters + ---------- + kind: + Profile kind (``"manual"`` or ``"tibber"``). + values: + Contract values dict as stored in ``EnergyContractVersion.values``. + + Returns + ------- + dict[str, Any] + A (possibly mutated) copy of *values* with defaults applied. + + Raises + ------ + ProfileNotFoundError + If the profile YAML for *kind* does not exist. + ProfileValidationError + If required fields are missing or have wrong types. + """ + profile = load_profile(kind) + if isinstance(profile, ManualProfile): + return _validate_manual_values(values, profile) + if isinstance(profile, TibberProfile): + return _validate_tibber_values(values, profile) + # Unreachable with current kinds, but guard for future extensions. + raise ProfileValidationError(f"No validator implemented for kind={kind!r}") diff --git a/app/integrations/pricing/profiles/manual.yaml b/app/integrations/pricing/profiles/manual.yaml new file mode 100644 index 0000000..c501a90 --- /dev/null +++ b/app/integrations/pricing/profiles/manual.yaml @@ -0,0 +1,20 @@ +kind: manual +label: 固定 / 可变费率（NL，双费率） + +energy: + dual_tariff: true # use delivered_1/2, returned_1/2 for low/high tariffs + buy: + normal: { unit: EUR/kWh } # high-tariff buy price (delivered_2 register) + dal: { unit: EUR/kWh } # low-tariff buy price (delivered_1 register) + sell: + normal: { unit: EUR/kWh } # high-tariff sell / return price + dal: { unit: EUR/kWh } # low-tariff sell / return price (currently equal to normal) + energy_tax: { unit: EUR/kWh } # energy tax added to buy price (incl. VAT) + ode: { unit: EUR/kWh, default: 0 } # currently merged into energy_tax + +standing: # fixed charges; UI fills per month, engine prorates to days + network_fee: { unit: EUR/month } + management_fee: { unit: EUR/month } + +credits: + heffingskorting: { unit: EUR/year } # energy-tax credit; deducted at summary layer diff --git a/app/integrations/pricing/profiles/tibber.yaml b/app/integrations/pricing/profiles/tibber.yaml new file mode 100644 index 0000000..4ec76f9 --- /dev/null +++ b/app/integrations/pricing/profiles/tibber.yaml @@ -0,0 +1,14 @@ +kind: tibber +label: Tibber 动态电价（15 分钟） + +energy: + source: tibber_api # buy = total (from API); sell = total − energy_tax − sell_adjust + energy_tax: { unit: EUR/kWh } # subtracted from total to derive sell price (incl. VAT) + sell_adjust: { unit: EUR/kWh, default: 0 } # additional sell-price adjustment (residual spread) + +standing: # fixed charges; UI fills per month, engine prorates to days + management_fee: { unit: EUR/month, default: 5.99 } + network_fee: { unit: EUR/month } + +credits: + heffingskorting: { unit: EUR/year } # energy-tax credit; deducted at summary layer diff --git a/app/integrations/pricing/strategies.py b/app/integrations/pricing/strategies.py new file mode 100644 index 0000000..f8b0f33 --- /dev/null +++ b/app/integrations/pricing/strategies.py @@ -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) diff --git a/docs/design/m6-tibber-dynamic-energy.md b/docs/design/m6-tibber-dynamic-energy.md index 23b775c..0b4e2f0 100644 --- a/docs/design/m6-tibber-dynamic-energy.md +++ b/docs/design/m6-tibber-dynamic-energy.md @@ -315,7 +315,7 @@ Phase D（API + 前端） - **Reviewer checklist**: secret 不回显/不进 OpenAPI 示例；`_settings_payload` 无遗漏；默认 off。 ### M6-T03 — pricing profile 框架 + strategy 注册表 -- **Status**: `todo` · **Depends**: none +- **Status**: `done` · **Depends**: none - **Context**: 仓库内 `manual.yaml`/`tibber.yaml` 定结构（pydantic 校验），加 price strategy 注册表（manual/tibber 出价）。纯模块，单测。 - **Files**: `create app/integrations/pricing/__init__.py`、`pricing/profiles.py`（pydantic 模型 + `load_profile`/`list_profiles`/`validate_values`）、`pricing/profiles/manual.yaml`、`pricing/profiles/tibber.yaml`、`pricing/strategies.py`（`register_strategy`/`get_strategy`、`manual`/`tibber` 实现）；`create tests/test_pricing_profiles.py`、`tests/test_pricing_strategies.py` - **Steps**: diff --git a/tests/test_pricing_profiles.py b/tests/test_pricing_profiles.py new file mode 100644 index 0000000..a5e7a2b --- /dev/null +++ b/tests/test_pricing_profiles.py @@ -0,0 +1,415 @@ +"""Tests for app/integrations/pricing/profiles.py. + +Acceptance criteria covered +---------------------------- +1. ``load_profile("manual")`` succeeds and returns a valid ``ManualProfile``. +2. ``load_profile("tibber")`` succeeds and returns a valid ``TibberProfile``. +3. Missing profile file raises ``ProfileNotFoundError``. +4. Malformed YAML (missing required fields) raises ``ProfileValidationError``. +5. Wrong kind in YAML raises ``ProfileValidationError``. +6. ``validate_values`` accepts conforming values for both kinds. +7. ``validate_values`` fills in default values (``ode``, ``sell_adjust``, + tibber ``management_fee``) when absent. +8. ``validate_values`` raises ``ProfileValidationError`` for missing required + fields (no default). +9. ``validate_values`` raises ``ProfileValidationError`` for wrong-typed fields. +10. ``list_profiles()`` returns both profiles in a list of dicts. +""" + +from __future__ import annotations + +import textwrap +from pathlib import Path +from unittest.mock import patch + +import pytest +import yaml + +from app.integrations.pricing.profiles import ( + ManualProfile, + ProfileNotFoundError, + ProfileValidationError, + TibberProfile, + list_profiles, + load_profile, + validate_values, +) + + +# --------------------------------------------------------------------------- +# 1-2: load_profile happy path +# --------------------------------------------------------------------------- + + +class TestLoadProfileManual: + """Validate that the shipped manual.yaml loads and validates correctly.""" + + def test_returns_manual_profile_instance(self) -> None: + profile = load_profile("manual") + assert isinstance(profile, ManualProfile) + + def test_kind_is_manual(self) -> None: + profile = load_profile("manual") + assert profile.kind == "manual" + + def test_has_label(self) -> None: + profile = load_profile("manual") + assert isinstance(profile.label, str) and profile.label + + def test_dual_tariff_is_true(self) -> None: + profile = load_profile("manual") + assert profile.energy.dual_tariff is True + + def test_energy_buy_has_normal_and_dal(self) -> None: + profile = load_profile("manual") + assert profile.energy.buy.normal.unit == "EUR/kWh" + assert profile.energy.buy.dal.unit == "EUR/kWh" + + def test_energy_sell_has_normal_and_dal(self) -> None: + profile = load_profile("manual") + assert profile.energy.sell.normal.unit == "EUR/kWh" + assert profile.energy.sell.dal.unit == "EUR/kWh" + + def test_energy_tax_unit(self) -> None: + profile = load_profile("manual") + assert profile.energy.energy_tax.unit == "EUR/kWh" + assert profile.energy.energy_tax.default is None # required field, no default + + def test_ode_has_default_zero(self) -> None: + profile = load_profile("manual") + assert profile.energy.ode.default == 0 + + def test_standing_fields(self) -> None: + profile = load_profile("manual") + assert profile.standing.network_fee.unit == "EUR/month" + assert profile.standing.management_fee.unit == "EUR/month" + + def test_credits_heffingskorting(self) -> None: + profile = load_profile("manual") + assert profile.credits.heffingskorting.unit == "EUR/year" + + def test_no_concrete_values_in_profile(self) -> None: + """Profile YAML must not contain any concrete price numbers.""" + profile_path = ( + Path(__file__).parent.parent + / "app/integrations/pricing/profiles/manual.yaml" + ) + with profile_path.open() as fh: + raw = yaml.safe_load(fh) + # Leaf nodes should only have 'unit' and optionally 'default: 0', + # not actual price values like 0.133 or 0.127. + energy = raw.get("energy", {}) + buy = energy.get("buy", {}) + # Leaf buy nodes: only 'unit' key, no numeric value key. + assert set(buy["normal"].keys()) == {"unit"}, ( + "buy.normal leaf must only have 'unit'" + ) + assert set(buy["dal"].keys()) == {"unit"}, ( + "buy.dal leaf must only have 'unit'" + ) + + +class TestLoadProfileTibber: + """Validate that the shipped tibber.yaml loads and validates correctly.""" + + def test_returns_tibber_profile_instance(self) -> None: + profile = load_profile("tibber") + assert isinstance(profile, TibberProfile) + + def test_kind_is_tibber(self) -> None: + profile = load_profile("tibber") + assert profile.kind == "tibber" + + def test_has_label(self) -> None: + profile = load_profile("tibber") + assert isinstance(profile.label, str) and profile.label + + def test_energy_source_is_tibber_api(self) -> None: + profile = load_profile("tibber") + assert profile.energy.source == "tibber_api" + + def test_energy_tax_unit(self) -> None: + profile = load_profile("tibber") + assert profile.energy.energy_tax.unit == "EUR/kWh" + assert profile.energy.energy_tax.default is None # required + + def test_sell_adjust_has_default_zero(self) -> None: + profile = load_profile("tibber") + assert profile.energy.sell_adjust.default == 0 + + def test_management_fee_has_default(self) -> None: + profile = load_profile("tibber") + assert profile.standing.management_fee.default is not None + assert isinstance(profile.standing.management_fee.default, float) + + def test_network_fee_unit(self) -> None: + profile = load_profile("tibber") + assert profile.standing.network_fee.unit == "EUR/month" + + def test_credits_heffingskorting(self) -> None: + profile = load_profile("tibber") + assert profile.credits.heffingskorting.unit == "EUR/year" + + +# --------------------------------------------------------------------------- +# 3-5: load_profile error cases +# --------------------------------------------------------------------------- + + +class TestLoadProfileErrors: + def test_missing_profile_raises_not_found(self) -> None: + with pytest.raises(ProfileNotFoundError, match="nonexistent"): + load_profile("nonexistent") + + def test_missing_required_fields_raises_validation_error( + self, tmp_path: Path + ) -> None: + """A YAML missing required fields raises ProfileValidationError.""" + bad_yaml = textwrap.dedent( + """\ + kind: manual + label: Bad manual profile + # missing energy / standing / credits sections entirely + """ + ) + bad_path = tmp_path / "manual.yaml" + bad_path.write_text(bad_yaml) + + with patch("app.integrations.pricing.profiles._PROFILES_DIR", tmp_path): + with pytest.raises(ProfileValidationError, match="manual"): + load_profile("manual") + + def test_wrong_type_raises_validation_error(self, tmp_path: Path) -> None: + """A YAML with a wrong type (string where dict is expected) raises ProfileValidationError.""" + bad_yaml = textwrap.dedent( + """\ + kind: tibber + label: Wrong type profile + energy: "should be a mapping not a string" + standing: + management_fee: { unit: EUR/month, default: 5.99 } + network_fee: { unit: EUR/month } + credits: + heffingskorting: { unit: EUR/year } + """ + ) + bad_path = tmp_path / "tibber.yaml" + bad_path.write_text(bad_yaml) + + with patch("app.integrations.pricing.profiles._PROFILES_DIR", tmp_path): + with pytest.raises(ProfileValidationError, match="tibber"): + load_profile("tibber") + + def test_unknown_kind_raises_validation_error(self, tmp_path: Path) -> None: + """A YAML with an unknown kind raises ProfileValidationError.""" + bad_yaml = textwrap.dedent( + """\ + kind: unknown_kind + label: Unknown kind profile + """ + ) + bad_path = tmp_path / "unknown_kind.yaml" + bad_path.write_text(bad_yaml) + + with patch("app.integrations.pricing.profiles._PROFILES_DIR", tmp_path): + with pytest.raises(ProfileValidationError, match="unknown_kind"): + load_profile("unknown_kind") + + +# --------------------------------------------------------------------------- +# 6-9: validate_values +# --------------------------------------------------------------------------- + +# A complete, conforming set of manual contract values. +_VALID_MANUAL_VALUES = { + "energy": { + "buy": {"normal": 0.133, "dal": 0.127}, + "sell": {"normal": 0.05, "dal": 0.05}, + "energy_tax": 0.1108, + "ode": 0.0, + }, + "standing": { + "network_fee": 9.87, + "management_fee": 9.87, + }, + "credits": { + "heffingskorting": 600.0, + }, +} + +# A complete, conforming set of tibber contract values. +_VALID_TIBBER_VALUES = { + "energy": { + "energy_tax": 0.1108, + "sell_adjust": 0.0, + }, + "standing": { + "management_fee": 5.99, + "network_fee": 9.87, + }, + "credits": { + "heffingskorting": 600.0, + }, +} + + +class TestValidateValuesManual: + def test_valid_values_accepted(self) -> None: + filled = validate_values("manual", dict(_VALID_MANUAL_VALUES)) + # Should not raise and should return a dict. + assert isinstance(filled, dict) + + def test_ode_default_applied_when_absent(self) -> None: + """When 'ode' is not in values, the default (0) should be inserted.""" + values = { + "energy": { + "buy": {"normal": 0.133, "dal": 0.127}, + "sell": {"normal": 0.05, "dal": 0.05}, + "energy_tax": 0.1108, + # ode is absent + }, + "standing": { + "network_fee": 9.87, + "management_fee": 9.87, + }, + "credits": {"heffingskorting": 600.0}, + } + filled = validate_values("manual", values) + assert filled["energy"]["ode"] == 0 + + def test_missing_energy_tax_raises(self) -> None: + values = { + "energy": { + "buy": {"normal": 0.133, "dal": 0.127}, + "sell": {"normal": 0.05, "dal": 0.05}, + # energy_tax absent — no default + }, + "standing": {"network_fee": 9.87, "management_fee": 9.87}, + "credits": {"heffingskorting": 600.0}, + } + with pytest.raises(ProfileValidationError, match="energy_tax"): + validate_values("manual", values) + + def test_missing_buy_normal_raises(self) -> None: + values = { + "energy": { + "buy": {"dal": 0.127}, # normal absent + "sell": {"normal": 0.05, "dal": 0.05}, + "energy_tax": 0.1108, + "ode": 0.0, + }, + "standing": {"network_fee": 9.87, "management_fee": 9.87}, + "credits": {"heffingskorting": 600.0}, + } + with pytest.raises(ProfileValidationError, match="normal"): + validate_values("manual", values) + + def test_wrong_type_for_energy_tax_raises(self) -> None: + values = { + "energy": { + "buy": {"normal": 0.133, "dal": 0.127}, + "sell": {"normal": 0.05, "dal": 0.05}, + "energy_tax": "not_a_number", # wrong type + "ode": 0.0, + }, + "standing": {"network_fee": 9.87, "management_fee": 9.87}, + "credits": {"heffingskorting": 600.0}, + } + with pytest.raises(ProfileValidationError, match="energy_tax"): + validate_values("manual", values) + + def test_missing_heffingskorting_raises(self) -> None: + values = { + "energy": { + "buy": {"normal": 0.133, "dal": 0.127}, + "sell": {"normal": 0.05, "dal": 0.05}, + "energy_tax": 0.1108, + "ode": 0.0, + }, + "standing": {"network_fee": 9.87, "management_fee": 9.87}, + "credits": {}, # heffingskorting absent — no default + } + with pytest.raises(ProfileValidationError, match="heffingskorting"): + validate_values("manual", values) + + +class TestValidateValuesTibber: + def test_valid_values_accepted(self) -> None: + filled = validate_values("tibber", dict(_VALID_TIBBER_VALUES)) + assert isinstance(filled, dict) + + def test_sell_adjust_default_applied_when_absent(self) -> None: + values = { + "energy": { + "energy_tax": 0.1108, + # sell_adjust absent — has default 0 + }, + "standing": {"management_fee": 5.99, "network_fee": 9.87}, + "credits": {"heffingskorting": 600.0}, + } + filled = validate_values("tibber", values) + assert filled["energy"]["sell_adjust"] == 0 + + def test_management_fee_default_applied_when_absent(self) -> None: + values = { + "energy": {"energy_tax": 0.1108, "sell_adjust": 0.0}, + "standing": { + "network_fee": 9.87, + # management_fee absent — has a default + }, + "credits": {"heffingskorting": 600.0}, + } + filled = validate_values("tibber", values) + assert "management_fee" in filled["standing"] + assert isinstance(filled["standing"]["management_fee"], float) + + def test_missing_energy_tax_raises(self) -> None: + values = { + "energy": {"sell_adjust": 0.0}, # energy_tax absent — no default + "standing": {"management_fee": 5.99, "network_fee": 9.87}, + "credits": {"heffingskorting": 600.0}, + } + with pytest.raises(ProfileValidationError, match="energy_tax"): + validate_values("tibber", values) + + def test_missing_network_fee_raises(self) -> None: + values = { + "energy": {"energy_tax": 0.1108, "sell_adjust": 0.0}, + "standing": {"management_fee": 5.99}, # network_fee absent — no default + "credits": {"heffingskorting": 600.0}, + } + with pytest.raises(ProfileValidationError, match="network_fee"): + validate_values("tibber", values) + + def test_wrong_type_for_sell_adjust_raises(self) -> None: + values = { + "energy": {"energy_tax": 0.1108, "sell_adjust": "zero"}, # wrong type + "standing": {"management_fee": 5.99, "network_fee": 9.87}, + "credits": {"heffingskorting": 600.0}, + } + with pytest.raises(ProfileValidationError, match="sell_adjust"): + validate_values("tibber", values) + + +# --------------------------------------------------------------------------- +# 10: list_profiles +# --------------------------------------------------------------------------- + + +class TestListProfiles: + def test_returns_list_of_dicts(self) -> None: + profiles = list_profiles() + assert isinstance(profiles, list) + for item in profiles: + assert isinstance(item, dict) + + def test_contains_manual_and_tibber(self) -> None: + profiles = list_profiles() + kinds = {p["kind"] for p in profiles} + assert "manual" in kinds + assert "tibber" in kinds + + def test_each_entry_has_label(self) -> None: + profiles = list_profiles() + for p in profiles: + assert "label" in p and isinstance(p["label"], str) diff --git a/tests/test_pricing_strategies.py b/tests/test_pricing_strategies.py new file mode 100644 index 0000000..be994f2 --- /dev/null +++ b/tests/test_pricing_strategies.py @@ -0,0 +1,553 @@ +"""Tests for app/integrations/pricing/strategies.py. + +Acceptance criteria covered +---------------------------- +1. ``get_strategy("manual")`` / ``get_strategy("tibber")`` return registered callables. +2. Manual strategy: dual-tariff import/export/net calculated correctly (hand-verified). +3. Manual strategy: Decimal precision — no float binary rounding errors. +4. Tibber strategy: queries the most recent TibberPrice with starts_at ≤ t0. +5. Tibber strategy: buy=total, sell=total−energy_tax−sell_adjust. +6. Tibber strategy: negative total → negative export_revenue (not clamped). +7. Tibber strategy: raises TibberPriceNotFoundError when no matching row exists. +8. ``register_strategy`` / ``get_strategy`` round-trip works. +9. ``get_strategy`` raises KeyError for unknown kinds. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from decimal import Decimal +from pathlib import Path + +import pytest +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from app.integrations.pricing.strategies import ( + PeriodDeltas, + TibberPriceNotFoundError, + get_strategy, + register_strategy, +) +from app.models.energy import TibberPrice + + +# --------------------------------------------------------------------------- +# Fixtures: in-memory / temp-file SQLite with energy tables +# --------------------------------------------------------------------------- + + +def _make_app_alembic_config(database_url: str) -> Config: + cfg = Config("alembic_app.ini") + cfg.set_main_option("sqlalchemy.url", database_url) + return cfg + + +@pytest.fixture() +def tibber_db(tmp_path: Path): + """Temporary SQLite DB upgraded to head (has tibber_price table).""" + db_path = tmp_path / "tibber_strategy_test.db" + db_url = f"sqlite:///{db_path}" + alembic_cfg = _make_app_alembic_config(db_url) + command.upgrade(alembic_cfg, "head") + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + yield engine + engine.dispose() + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_UTC = timezone.utc + + +def _ts(hour: int, minute: int = 0) -> datetime: + """Return a UTC datetime on 2026-06-23 at the given hour:minute.""" + return datetime(2026, 6, 23, hour, minute, 0, tzinfo=_UTC) + + +def _insert_tibber_price( + session: Session, + starts_at: datetime, + total: float, + energy: float = 0.08, + tax: float | None = None, + currency: str = "EUR", +) -> TibberPrice: + """Insert and flush a TibberPrice row; return the ORM instance.""" + if tax is None: + tax = round(total - energy, 6) + row = TibberPrice( + starts_at=starts_at, + resolution="QUARTER_HOURLY", + energy=energy, + tax=tax, + total=total, + level="NORMAL", + currency=currency, + fetched_at=datetime.now(_UTC), + ) + session.add(row) + session.flush() + return row + + +# --------------------------------------------------------------------------- +# 1. Registry round-trip +# --------------------------------------------------------------------------- + + +class TestRegistry: + def test_manual_strategy_registered(self) -> None: + fn = get_strategy("manual") + assert callable(fn) + + def test_tibber_strategy_registered(self) -> None: + fn = get_strategy("tibber") + assert callable(fn) + + def test_unknown_kind_raises_key_error(self) -> None: + with pytest.raises(KeyError, match="no_such_kind"): + get_strategy("no_such_kind") + + def test_register_and_retrieve_custom_strategy(self) -> None: + def _my_fn(deltas, t0, values, session): + return {} + + register_strategy("_test_custom", _my_fn) + assert get_strategy("_test_custom") is _my_fn + # Cleanup to avoid polluting other tests. + from app.integrations.pricing.strategies import _REGISTRY + _REGISTRY.pop("_test_custom", None) + + +# --------------------------------------------------------------------------- +# 2-3. Manual strategy +# --------------------------------------------------------------------------- + +# Dummy session (manual strategy does not use the session). +_NO_SESSION = None # type: ignore[assignment] + + +class TestManualStrategy: + """Verify manual dual-tariff price calculations.""" + + # Contract values for all manual tests. + _VALUES = { + "energy": { + "buy": {"normal": 0.133, "dal": 0.127}, + "sell": {"normal": 0.05, "dal": 0.05}, + "energy_tax": 0.11, + "ode": 0.0, + }, + "standing": {"network_fee": 9.87, "management_fee": 9.87}, + "credits": {"heffingskorting": 600.0}, + } + + def _call(self, deltas: PeriodDeltas) -> dict: + fn = get_strategy("manual") + return fn(deltas, _ts(10), self._VALUES, _NO_SESSION) + + # --- hand-calculated expected values --- + # buy_dal = 0.127 + 0.11 + 0.0 = 0.237 + # buy_normal = 0.133 + 0.11 + 0.0 = 0.243 + # sell_dal = 0.05 + # sell_normal = 0.05 + # + # With Δd1=2, Δd2=3, Δr1=1, Δr2=4: + # import_cost = 2×0.237 + 3×0.243 = 0.474 + 0.729 = 1.203 + # export_revenue = 1×0.05 + 4×0.05 = 0.05 + 0.20 = 0.25 + # net_cost = 1.203 − 0.25 = 0.953 + + def test_import_cost_dual_tariff(self) -> None: + deltas = PeriodDeltas( + d1=Decimal("2"), d2=Decimal("3"), + r1=Decimal("1"), r2=Decimal("4"), + ) + result = self._call(deltas) + expected = Decimal("2") * Decimal("0.237") + Decimal("3") * Decimal("0.243") + assert result["import_cost"] == expected + + def test_export_revenue_dual_tariff(self) -> None: + deltas = PeriodDeltas( + d1=Decimal("2"), d2=Decimal("3"), + r1=Decimal("1"), r2=Decimal("4"), + ) + result = self._call(deltas) + expected = Decimal("1") * Decimal("0.05") + Decimal("4") * Decimal("0.05") + assert result["export_revenue"] == expected + + def test_net_cost_equals_import_minus_export(self) -> None: + deltas = PeriodDeltas( + d1=Decimal("2"), d2=Decimal("3"), + r1=Decimal("1"), r2=Decimal("4"), + ) + result = self._call(deltas) + assert result["net_cost"] == result["import_cost"] - result["export_revenue"] + + def test_hand_calculated_values(self) -> None: + """Verify exact hand-calculated result for the reference deltas.""" + deltas = PeriodDeltas( + d1=Decimal("2"), d2=Decimal("3"), + r1=Decimal("1"), r2=Decimal("4"), + ) + result = self._call(deltas) + assert result["import_cost"] == Decimal("1.203") + assert result["export_revenue"] == Decimal("0.25") + assert result["net_cost"] == Decimal("0.953") + + def test_zero_deltas_yields_zero_costs(self) -> None: + deltas = PeriodDeltas( + d1=Decimal("0"), d2=Decimal("0"), + r1=Decimal("0"), r2=Decimal("0"), + ) + result = self._call(deltas) + assert result["import_cost"] == Decimal("0") + assert result["export_revenue"] == Decimal("0") + assert result["net_cost"] == Decimal("0") + + def test_ode_included_in_buy_price(self) -> None: + """When ode > 0 it is added to the effective buy price.""" + values = { + "energy": { + "buy": {"normal": 0.10, "dal": 0.10}, + "sell": {"normal": 0.05, "dal": 0.05}, + "energy_tax": 0.10, + "ode": 0.01, # non-zero ode + }, + "standing": {"network_fee": 0.0, "management_fee": 0.0}, + "credits": {"heffingskorting": 0.0}, + } + deltas = PeriodDeltas( + d1=Decimal("1"), d2=Decimal("0"), + r1=Decimal("0"), r2=Decimal("0"), + ) + fn = get_strategy("manual") + result = fn(deltas, _ts(10), values, _NO_SESSION) + # buy_dal = 0.10 + 0.10 + 0.01 = 0.21; import_cost = 1 × 0.21 = 0.21 + assert result["import_cost"] == Decimal("0.21") + + def test_import_export_kept_separate(self) -> None: + """import_cost and export_revenue must not be netted before assignment.""" + deltas = PeriodDeltas( + d1=Decimal("5"), d2=Decimal("5"), + r1=Decimal("3"), r2=Decimal("3"), + ) + result = self._call(deltas) + # Both must be individually non-zero. + assert result["import_cost"] > 0 + assert result["export_revenue"] > 0 + + def test_pricing_snapshot_present(self) -> None: + deltas = PeriodDeltas( + d1=Decimal("1"), d2=Decimal("1"), + r1=Decimal("0"), r2=Decimal("0"), + ) + result = self._call(deltas) + snapshot = result["pricing"] + assert snapshot["kind"] == "manual" + assert "buy_dal" in snapshot + assert "buy_normal" in snapshot + assert "sell_dal" in snapshot + assert "sell_normal" in snapshot + + +# --------------------------------------------------------------------------- +# 3. Decimal precision — float-error exposure test +# --------------------------------------------------------------------------- + + +class TestManualDecimalPrecision: + """Use values known to produce binary float errors if float arithmetic is used.""" + + def test_no_float_rounding_error(self) -> None: + """0.1 + 0.2 in float gives 0.30000000000000004; Decimal must be exact.""" + values = { + "energy": { + "buy": {"normal": 0.1, "dal": 0.2}, + "sell": {"normal": 0.1, "dal": 0.1}, + "energy_tax": 0.0, + "ode": 0.0, + }, + "standing": {"network_fee": 0.0, "management_fee": 0.0}, + "credits": {"heffingskorting": 0.0}, + } + deltas = PeriodDeltas( + d1=Decimal("1"), d2=Decimal("1"), + r1=Decimal("0"), r2=Decimal("0"), + ) + fn = get_strategy("manual") + result = fn(deltas, _ts(10), values, _NO_SESSION) + # import_cost = 1×0.2 + 1×0.1 = 0.3 exactly + assert result["import_cost"] == Decimal("0.3"), ( + f"Expected Decimal('0.3'), got {result['import_cost']!r} — " + "float arithmetic leaking in?" + ) + + def test_result_values_are_decimal_type(self) -> None: + values = { + "energy": { + "buy": {"normal": 0.133, "dal": 0.127}, + "sell": {"normal": 0.05, "dal": 0.05}, + "energy_tax": 0.11, + "ode": 0.0, + }, + "standing": {"network_fee": 9.87, "management_fee": 9.87}, + "credits": {"heffingskorting": 600.0}, + } + deltas = PeriodDeltas( + d1=Decimal("1"), d2=Decimal("1"), + r1=Decimal("1"), r2=Decimal("1"), + ) + fn = get_strategy("manual") + result = fn(deltas, _ts(10), values, _NO_SESSION) + assert isinstance(result["import_cost"], Decimal) + assert isinstance(result["export_revenue"], Decimal) + assert isinstance(result["net_cost"], Decimal) + + +# --------------------------------------------------------------------------- +# 4-7. Tibber strategy +# --------------------------------------------------------------------------- + + +class TestTibberStrategy: + """Verify Tibber price-lookup and billing calculations.""" + + _VALUES = { + "energy": { + "energy_tax": 0.10, + "sell_adjust": 0.0, + }, + "standing": {"management_fee": 5.99, "network_fee": 9.87}, + "credits": {"heffingskorting": 600.0}, + } + + def _call( + self, + deltas: PeriodDeltas, + t0: datetime, + session: Session, + values: dict | None = None, + ) -> dict: + fn = get_strategy("tibber") + return fn(deltas, t0, values or self._VALUES, session) + + def test_buy_equals_total(self, tibber_db) -> None: + t0 = _ts(10, 0) + with Session(tibber_db) as session: + _insert_tibber_price(session, starts_at=_ts(9, 45), total=0.25) + session.commit() + + with Session(tibber_db) as session: + deltas = PeriodDeltas( + d1=Decimal("1"), d2=Decimal("1"), + r1=Decimal("0"), r2=Decimal("0"), + ) + result = self._call(deltas, t0, session) + # buy = total = 0.25; import_cost = 2 × 0.25 = 0.50 + assert result["import_cost"] == Decimal("2") * Decimal("0.25") + + def test_sell_equals_total_minus_energy_tax_minus_adjust(self, tibber_db) -> None: + t0 = _ts(10, 0) + with Session(tibber_db) as session: + _insert_tibber_price(session, starts_at=_ts(9, 45), total=0.25) + session.commit() + + values = { + "energy": {"energy_tax": 0.10, "sell_adjust": 0.02}, + "standing": {"management_fee": 5.99, "network_fee": 9.87}, + "credits": {"heffingskorting": 600.0}, + } + with Session(tibber_db) as session: + deltas = PeriodDeltas( + d1=Decimal("0"), d2=Decimal("0"), + r1=Decimal("1"), r2=Decimal("1"), + ) + result = self._call(deltas, t0, session, values=values) + # sell = 0.25 - 0.10 - 0.02 = 0.13; export_revenue = 2 × 0.13 = 0.26 + assert result["export_revenue"] == Decimal("2") * Decimal("0.13") + + def test_uses_most_recent_price_before_t0(self, tibber_db) -> None: + """Correct row: starts_at ≤ t0, most recent wins.""" + t0 = _ts(10, 0) + with Session(tibber_db) as session: + # Older price (should NOT be used) + _insert_tibber_price(session, starts_at=_ts(9, 0), total=0.10) + # Closer price (starts_at ≤ t0, should be used) + _insert_tibber_price(session, starts_at=_ts(9, 45), total=0.30) + # Future price (starts_at > t0, must NOT be used) + _insert_tibber_price(session, starts_at=_ts(10, 15), total=0.99) + session.commit() + + with Session(tibber_db) as session: + deltas = PeriodDeltas( + d1=Decimal("1"), d2=Decimal("0"), + r1=Decimal("0"), r2=Decimal("0"), + ) + result = self._call(deltas, t0, session) + # Only the 09:45 price (total=0.30) should be used. + snapshot = result["pricing"] + assert Decimal(snapshot["total"]) == Decimal("0.30") + + def test_tibber_sums_both_tariff_registers(self, tibber_db) -> None: + """Tibber does not split dal/normal; import_cost = (d1+d2) × buy.""" + t0 = _ts(10, 0) + with Session(tibber_db) as session: + _insert_tibber_price(session, starts_at=_ts(9, 45), total=0.20) + session.commit() + + with Session(tibber_db) as session: + deltas = PeriodDeltas( + d1=Decimal("3"), d2=Decimal("2"), + r1=Decimal("0"), r2=Decimal("0"), + ) + result = self._call(deltas, t0, session) + # import_cost = (3+2) × 0.20 = 1.00 + assert result["import_cost"] == Decimal("1.00") + + def test_negative_total_gives_negative_export_revenue(self, tibber_db) -> None: + """When total is negative, selling electricity costs money (correct behaviour).""" + t0 = _ts(10, 0) + with Session(tibber_db) as session: + # total = -0.05; sell = -0.05 - 0.10 - 0.0 = -0.15 + _insert_tibber_price(session, starts_at=_ts(9, 45), total=-0.05) + session.commit() + + with Session(tibber_db) as session: + deltas = PeriodDeltas( + d1=Decimal("0"), d2=Decimal("0"), + r1=Decimal("2"), r2=Decimal("2"), + ) + result = self._call(deltas, t0, session) + # sell = -0.05 - 0.10 - 0.0 = -0.15; export_revenue = 4 × (-0.15) = -0.60 + assert result["export_revenue"] < 0 + assert result["export_revenue"] == Decimal("4") * ( + Decimal("-0.05") - Decimal("0.10") - Decimal("0.0") + ) + + def test_negative_total_exact_calculation(self, tibber_db) -> None: + """Full hand-calculation for negative-total scenario.""" + t0 = _ts(10, 0) + total = Decimal("-0.05") + energy_tax = Decimal("0.10") + sell_adjust = Decimal("0.0") + with Session(tibber_db) as session: + _insert_tibber_price(session, starts_at=_ts(9, 45), total=float(total)) + session.commit() + + with Session(tibber_db) as session: + deltas = PeriodDeltas( + d1=Decimal("1"), d2=Decimal("1"), # delivered = 2 kWh + r1=Decimal("1"), r2=Decimal("1"), # returned = 2 kWh + ) + result = self._call(deltas, t0, session) + + buy = total # -0.05 + sell = total - energy_tax - sell_adjust # -0.15 + expected_import = Decimal("2") * buy # -0.10 + expected_export = Decimal("2") * sell # -0.30 + expected_net = expected_import - expected_export # 0.20 + + assert result["import_cost"] == expected_import + assert result["export_revenue"] == expected_export + assert result["net_cost"] == expected_net + + def test_no_price_before_t0_raises(self, tibber_db) -> None: + """When no TibberPrice exists with starts_at ≤ t0, TibberPriceNotFoundError is raised.""" + t0 = _ts(10, 0) + with Session(tibber_db) as session: + # Only a future price — starts_at > t0. + _insert_tibber_price(session, starts_at=_ts(10, 15), total=0.25) + session.commit() + + fn = get_strategy("tibber") + with Session(tibber_db) as session: + deltas = PeriodDeltas( + d1=Decimal("1"), d2=Decimal("0"), + r1=Decimal("0"), r2=Decimal("0"), + ) + with pytest.raises(TibberPriceNotFoundError): + fn(deltas, t0, self._VALUES, session) + + def test_empty_tibber_table_raises(self, tibber_db) -> None: + """Empty tibber_price table raises TibberPriceNotFoundError.""" + fn = get_strategy("tibber") + with Session(tibber_db) as session: + deltas = PeriodDeltas( + d1=Decimal("1"), d2=Decimal("0"), + r1=Decimal("0"), r2=Decimal("0"), + ) + with pytest.raises(TibberPriceNotFoundError): + fn(deltas, _ts(10), self._VALUES, session) + + def test_pricing_snapshot_contains_expected_keys(self, tibber_db) -> None: + t0 = _ts(10, 0) + with Session(tibber_db) as session: + _insert_tibber_price(session, starts_at=_ts(9, 45), total=0.20) + session.commit() + + with Session(tibber_db) as session: + deltas = PeriodDeltas( + d1=Decimal("1"), d2=Decimal("0"), + r1=Decimal("0"), r2=Decimal("0"), + ) + result = self._call(deltas, t0, session) + snapshot = result["pricing"] + assert snapshot["kind"] == "tibber" + assert "tibber_price_starts_at" in snapshot + assert "buy" in snapshot + assert "sell" in snapshot + + def test_result_values_are_decimal_type(self, tibber_db) -> None: + t0 = _ts(10, 0) + with Session(tibber_db) as session: + _insert_tibber_price(session, starts_at=_ts(9, 45), total=0.20) + session.commit() + + with Session(tibber_db) as session: + deltas = PeriodDeltas( + d1=Decimal("1"), d2=Decimal("1"), + r1=Decimal("1"), r2=Decimal("1"), + ) + result = self._call(deltas, t0, session) + assert isinstance(result["import_cost"], Decimal) + assert isinstance(result["export_revenue"], Decimal) + assert isinstance(result["net_cost"], Decimal) + + +# --------------------------------------------------------------------------- +# Tibber precision test +# --------------------------------------------------------------------------- + + +class TestTibberDecimalPrecision: + """Ensure Tibber arithmetic is exact Decimal (no float leakage).""" + + def test_no_float_rounding_for_tricky_values(self, tibber_db) -> None: + """total=0.1, energy_tax=0.2 — float gives 0.1+0.2 error; Decimal must be exact.""" + t0 = _ts(10, 0) + with Session(tibber_db) as session: + _insert_tibber_price(session, starts_at=_ts(9, 45), total=0.3) + session.commit() + + values = { + "energy": {"energy_tax": 0.1, "sell_adjust": 0.2}, + "standing": {"management_fee": 5.99, "network_fee": 9.87}, + "credits": {"heffingskorting": 600.0}, + } + fn = get_strategy("tibber") + with Session(tibber_db) as session: + deltas = PeriodDeltas( + d1=Decimal("0"), d2=Decimal("0"), + r1=Decimal("1"), r2=Decimal("0"), + ) + result = fn(deltas, t0, values, session) + # sell = 0.3 - 0.1 - 0.2 = 0.0 exactly (float gives ~2.8e-17) + assert result["export_revenue"] == Decimal("0"), ( + f"Expected Decimal('0'), got {result['export_revenue']!r} — " + "float arithmetic leaking in?" + )