"""Billing engine for DSMR 15-minute energy metering periods. This module implements the two-layer billing model described in §3.4 of the M6 design document: **Layer 1 — per-period metering cost (immutable, price-snapshot)** ``compute_period(session, t0)`` computes the import cost, export revenue, and net cost for the 15-minute period ``[t0, t0+15min)``. The result is written to ``energy_cost_period`` with a full pricing snapshot so each row is self-contained and auditable. Existing *successful* rows are never overwritten by the normal tick path; only an explicit ``recompute_range`` call passes ``overwrite=True``. **Layer 2 — summary (computed at read time, not stored)** ``summarize(session, start, end)`` aggregates all non-degraded ``energy_cost_period`` rows in ``[start, end)``, then adds the daily standing charges (network_fee + management_fee, apportioned at EUR/month ÷ 30 per day) and subtracts the energy-tax credit (heffingskorting, apportioned at EUR/year ÷ 365 per day). Design notes ------------ - **Decimal arithmetic throughout**: all monetary computations use ``decimal.Decimal`` to avoid float binary rounding errors. Only when writing to ``EnergyCostPeriod`` columns (Float) are values converted to float. ``summarize`` converts back to Decimal for summation. - **UTC quarter-hour grid**: period boundaries are aligned to UTC 00/15/30/45 minutes (``floor_to_quarter``). NL local time (CET/CEST) is always a whole number of hours from UTC, so the quarter-hour grid is the same in both timezone representations. - **Register keys**: DSMR payload uses JSON strings like ``"20915.154"`` for cumulative kWh registers. ``register_at`` converts them to Decimal. - **Degraded vs skip semantics**: - *Missing readings* (``register_at`` returns None for start or end boundary): write a ``degraded=True`` row with costs at 0 so the period is tracked and can be retried by ``compute_closed_periods``. - *Missing Tibber price* (``TibberPriceNotFoundError``): skip entirely (do not write a row); the period will be retried once prices arrive. - *Missing active contract version*: skip (no contract to compute against). - **Lookback window in ``compute_closed_periods``**: to avoid scanning all historical DSMR data on every tick, the function looks back at most 7 days from the current time. This covers typical short outages (no data / no contract) while staying bounded. Periods older than 7 days must be recovered via an explicit ``recompute_range`` call. """ from __future__ import annotations import logging from datetime import UTC, datetime, timedelta from decimal import Decimal from typing import Any from sqlalchemy import select from sqlalchemy.orm import Session from app.integrations.pricing.strategies import ( PeriodDeltas, TibberPriceNotFoundError, get_strategy, ) from app.models.energy import DsmrReading, EnergyCostPeriod from app.services.contracts import active_contract_version_at, active_contract_versions from app.services.timezone import local_date, local_now logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- _PERIOD_MINUTES = 15 _LOOKBACK_DAYS = 7 # maximum lookback window for compute_closed_periods # Maximum age a DSMR reading may have relative to the boundary being queried. # Under normal operation DSMR readings arrive every ~10 seconds, so a reading # more than one period (15 minutes) old at the boundary indicates either a data # gap or — critically — a *future* boundary being resolved against the last # historical reading. In both cases the reading is considered stale and # ``register_at`` returns None, letting the period be marked degraded instead of # producing a spurious zero-delta "successful" row. _READING_MAX_STALENESS = timedelta(minutes=_PERIOD_MINUTES) # DSMR payload register keys (cumulative kWh, JSON string values). _KEY_D1 = "electricity_delivered_1" # delivered low-tariff (dal / _1) _KEY_D2 = "electricity_delivered_2" # delivered high-tariff (normal / _2) _KEY_R1 = "electricity_returned_1" # returned low-tariff _KEY_R2 = "electricity_returned_2" # returned high-tariff # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def floor_to_quarter(dt: datetime) -> datetime: """Return *dt* floored to the nearest UTC quarter-hour boundary. The result always has seconds=0 and microseconds=0, and minutes in {0, 15, 30, 45}. Timezone info is preserved if present. """ floored_minute = (dt.minute // _PERIOD_MINUTES) * _PERIOD_MINUTES return dt.replace(minute=floored_minute, second=0, microsecond=0) def _to_decimal(value: Any) -> Decimal: """Convert *value* to Decimal via str() to avoid float binary rounding.""" return Decimal(str(value)) def _as_utc(dt: datetime) -> datetime: """Attach UTC tzinfo to a naive datetime (SQLite read-back workaround).""" if dt.tzinfo is None: return dt.replace(tzinfo=UTC) return dt def _existing_period(session: Session, t0: datetime) -> EnergyCostPeriod | None: """Return the EnergyCostPeriod row for period_start=t0, or None.""" return session.execute( select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == t0) ).scalar_one_or_none() # --------------------------------------------------------------------------- # register_at — boundary reading lookup # --------------------------------------------------------------------------- def register_at(session: Session, boundary: datetime) -> dict[str, Decimal] | None: """Return the four cumulative kWh register values at *boundary*. Queries the most recent ``DsmrReading`` with ``recorded_at ≤ boundary`` and extracts the four energy registers from ``payload``: d1 — electricity_delivered_1 (delivered low-tariff / dal) d2 — electricity_delivered_2 (delivered high-tariff / normal) r1 — electricity_returned_1 (returned low-tariff) r2 — electricity_returned_2 (returned high-tariff) Returns ------- dict[str, Decimal] with keys ``d1``, ``d2``, ``r1``, ``r2``, or ``None`` when: - No ``DsmrReading`` row exists with ``recorded_at ≤ boundary``. - Any of the four register keys is absent from the payload. - Any of the four register values is ``None`` (null in JSON). """ row: DsmrReading | None = ( session.execute( select(DsmrReading) .where(DsmrReading.recorded_at <= boundary) .order_by(DsmrReading.recorded_at.desc()) .limit(1) ).scalar_one_or_none() ) if row is None: return None # Freshness guard: reject readings that are too old relative to *boundary*. # ``recorded_at`` is stored as a naive UTC datetime in SQLite; attach UTC # tzinfo before comparing with *boundary* (which is always tz-aware UTC) to # avoid an "offset-naive vs offset-aware" TypeError. if _as_utc(row.recorded_at) < _as_utc(boundary) - _READING_MAX_STALENESS: return None payload = row.payload or {} try: d1_raw = payload[_KEY_D1] d2_raw = payload[_KEY_D2] r1_raw = payload[_KEY_R1] r2_raw = payload[_KEY_R2] except KeyError: return None if any(v is None for v in (d1_raw, d2_raw, r1_raw, r2_raw)): return None return { "d1": _to_decimal(d1_raw), "d2": _to_decimal(d2_raw), "r1": _to_decimal(r1_raw), "r2": _to_decimal(r2_raw), } # --------------------------------------------------------------------------- # compute_period — single 15-minute period # --------------------------------------------------------------------------- def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -> bool: """Compute and upsert the billing record for the period ``[t0, t0+15min)``. Parameters ---------- session: Active SQLAlchemy session. Caller is responsible for committing. t0: UTC start of the 15-minute period. **Must** lie on a quarter-hour grid boundary (minutes ∈ {0, 15, 30, 45}, seconds=0, microseconds=0). overwrite: If ``True``, overwrite an existing *successful* row (i.e. re-compute even when a non-degraded record already exists). The normal tick path always passes ``False``; only ``recompute_range`` passes ``True``. Returns ------- bool ``True`` if a record was written (inserted or updated), ``False`` if the period was skipped (missing contract or missing Tibber price). Side-effects ------------ - Inserts or updates an ``EnergyCostPeriod`` row keyed on ``period_start=t0``. - If readings are missing at either boundary: inserts/updates a degraded row (costs=0, degraded=True). - If the active contract version is missing: **skips** (returns False, no write). - If the Tibber price is missing (TibberPriceNotFoundError): **skips** (returns False, no write). """ t1 = t0 + timedelta(minutes=_PERIOD_MINUTES) now = datetime.now(UTC) # Immutability guard: skip if a successful record already exists and we # are not in overwrite mode. existing = _existing_period(session, t0) if existing is not None and not existing.degraded and not overwrite: return False # --- Active contract version at t0 (checked before readings) --- # If there is no active contract covering t0, skip the period entirely. # We do not write a degraded row — there is no meaningful state to recover # without a contract (we would not know which strategy to apply once data # arrives). The period can be recovered via an explicit recompute_range once # a contract is configured and activated. version = active_contract_version_at(session, t0) if version is None: logger.debug("compute_period(%s): no active contract version — skipping.", t0.isoformat()) return False # --- Boundary readings --- start_regs = register_at(session, t0) end_regs = register_at(session, t1) if start_regs is None or end_regs is None: # Missing readings → write/update a degraded placeholder so the period # is visible and can be retried by compute_closed_periods once data arrives. _upsert_degraded(session, t0, now, existing) return True # a record was written (degraded) # --- Compute deltas (end − start) --- deltas = PeriodDeltas( d1=end_regs["d1"] - start_regs["d1"], d2=end_regs["d2"] - start_regs["d2"], r1=end_regs["r1"] - start_regs["r1"], r2=end_regs["r2"] - start_regs["r2"], ) # --- Price strategy --- strategy = get_strategy(version.contract.kind) try: result = strategy(deltas, t0, version.values, session) except TibberPriceNotFoundError: # Missing Tibber price → skip the period; it will be retried once the # price arrives (e.g. after the next Tibber refresh job runs). logger.debug( "compute_period(%s): no Tibber price found — skipping.", t0.isoformat() ) return False # --- Upsert the billing record --- import_cost: Decimal = result["import_cost"] export_revenue: Decimal = result["export_revenue"] net_cost: Decimal = result["net_cost"] pricing: dict = result["pricing"] if existing is not None: # Update in-place (overwrite=True or previous record was degraded). existing.d1_kwh = float(deltas.d1) existing.d2_kwh = float(deltas.d2) existing.r1_kwh = float(deltas.r1) existing.r2_kwh = float(deltas.r2) existing.import_cost = float(import_cost) existing.export_revenue = float(export_revenue) existing.net_cost = float(net_cost) existing.currency = version.contract.currency existing.pricing = pricing existing.contract_version_id = version.id existing.degraded = False existing.computed_at = now else: period = EnergyCostPeriod( period_start=t0, d1_kwh=float(deltas.d1), d2_kwh=float(deltas.d2), r1_kwh=float(deltas.r1), r2_kwh=float(deltas.r2), import_cost=float(import_cost), export_revenue=float(export_revenue), net_cost=float(net_cost), currency=version.contract.currency, pricing=pricing, contract_version_id=version.id, degraded=False, computed_at=now, ) session.add(period) return True def _upsert_degraded( session: Session, t0: datetime, now: datetime, existing: EnergyCostPeriod | None, ) -> None: """Insert or update a degraded placeholder for period *t0*. When *existing* is not None (row was previously written — either degraded or successful), the row is explicitly reset to the standard degraded state. This is required for the ``recompute_range`` (overwrite=True) path: if the row was previously a *successful* computation and the boundary readings have since disappeared, the stale non-zero costs must be cleared so the row accurately reflects the current "missing readings" state rather than masquerading as a valid result. """ if existing is not None: # Explicitly reset to degraded state — identical field values to the # new-row path below. This covers the recompute-over-successful-row # case where old non-zero costs must not survive the downgrade. existing.d1_kwh = 0.0 existing.d2_kwh = 0.0 existing.r1_kwh = 0.0 existing.r2_kwh = 0.0 existing.import_cost = 0.0 existing.export_revenue = 0.0 existing.net_cost = 0.0 existing.pricing = {} existing.contract_version_id = None existing.degraded = True existing.computed_at = now else: period = EnergyCostPeriod( period_start=t0, d1_kwh=0.0, d2_kwh=0.0, r1_kwh=0.0, r2_kwh=0.0, import_cost=0.0, export_revenue=0.0, net_cost=0.0, currency="EUR", # placeholder; real currency known after contract lookup pricing={}, contract_version_id=None, degraded=True, computed_at=now, ) session.add(period) # --------------------------------------------------------------------------- # compute_closed_periods — periodic tick # --------------------------------------------------------------------------- def compute_closed_periods(session: Session) -> int: """Find and compute all uncalculated closed 15-minute periods. A period ``[t0, t1)`` is *closed* when ``t1 ≤ now``. This function: 1. Determines the lookback window: from ``now − LOOKBACK_DAYS`` to ``now``, floored to the nearest quarter-hour. This avoids an unbounded full historical scan on every tick while still covering the typical recovery window (short outages, missing contract, etc.). Periods older than ``LOOKBACK_DAYS`` must be recovered via an explicit ``recompute_range``. 2. Iterates over all quarter-hour boundaries in that window where ``t1 ≤ now`` (i.e. the period has already closed). 3. For each boundary, calls ``compute_period(overwrite=False)``, which: - Skips periods that already have a *successful* (non-degraded) record. - Retries periods that have a *degraded* record. - Skips periods for which no active contract version exists or the Tibber price is unavailable (without writing a degraded row). Returns ------- int Number of periods for which a record was written (inserted or updated). Does not count skipped periods. """ now = datetime.now(UTC) # Current period boundary (the one whose t1 has not yet passed). current_t0 = floor_to_quarter(now) # Earliest boundary to consider. earliest_t0 = floor_to_quarter(now - timedelta(days=_LOOKBACK_DAYS)) written = 0 t0 = earliest_t0 while t0 < current_t0: t1 = t0 + timedelta(minutes=_PERIOD_MINUTES) if t1 <= now: try: did_write = compute_period(session, t0, overwrite=False) if did_write: written += 1 except Exception: logger.exception( "compute_closed_periods: unexpected error for t0=%s — continuing.", t0.isoformat(), ) t0 += timedelta(minutes=_PERIOD_MINUTES) if written: session.commit() logger.info("compute_closed_periods: wrote %d period(s).", written) return written # --------------------------------------------------------------------------- # recompute_range — explicit full recompute # --------------------------------------------------------------------------- def recompute_range(session: Session, start: datetime, end: datetime) -> int: """Recompute (overwrite) all 15-minute periods in ``[start, end)``. This is the *explicit opt-in* path for recovering from: - Periods where readings or prices arrived late. - Price corrections (new contract version retroactively applied). - Any other reason to override the immutability guard. The function iterates over every UTC quarter-hour boundary in ``[floor(start), end)`` and calls ``compute_period(overwrite=True)``. Existing rows (including successful ones) are overwritten. Parameters ---------- session: Active SQLAlchemy session. The function commits after all periods have been processed. start: Inclusive start datetime (floored to the nearest quarter-hour internally). end: Exclusive end datetime. Returns ------- int Number of periods for which a record was written (inserted or updated). Periods skipped due to missing contract or missing Tibber price are *not* counted. """ t0 = floor_to_quarter(_as_utc(start)) end_utc = _as_utc(end) now = datetime.now(UTC) written = 0 while t0 < end_utc: t1 = t0 + timedelta(minutes=_PERIOD_MINUTES) # Only recompute periods that have already closed (t1 ≤ now). Even # when the caller passes a future *end*, we must not write speculative # "future" rows: register_at would resolve to the latest historical # reading for both boundaries, producing a zero-delta fake-success row # that blocks the real computation once the period actually closes. if t1 <= now: try: did_write = compute_period(session, t0, overwrite=True) if did_write: written += 1 except Exception: logger.exception( "recompute_range: unexpected error for t0=%s — continuing.", t0.isoformat(), ) t0 += timedelta(minutes=_PERIOD_MINUTES) session.commit() logger.info( "recompute_range(%s, %s): wrote %d period(s).", start.isoformat(), end.isoformat(), written, ) return written # --------------------------------------------------------------------------- # summarize — layer-2 aggregation (read-time, not stored) # --------------------------------------------------------------------------- def summarize(session: Session, start: datetime, end: datetime) -> dict[str, Any]: """Aggregate billing for the interval ``[start, end)``. Computes the total payable as: total_payable = Σ(net_cost) -- metered electricity + fixed_costs -- per-day standing charges, cross-version - credits -- per-day heffingskorting, cross-version **Fixed-cost / credit counting — Principle C**: Only *already-elapsed* whole local calendar days are counted. For each local calendar date D in the window ``[start_local_date, min(end_local_date, tomorrow_local))``, the contract version whose rate covers D (the one whose effective_from local-date ≤ D < next version's effective_from local-date) is used. Days that have not yet started in local time (D > today_local) are never counted. This is cross-version: if the active contract has V1 from June 1 and V2 from June 25, querying June 1–30 uses V1 for days 1-24 and V2 for day 25. Switching versions never resets the counter. **Timezone note**: the ``days`` field in the returned dict still represents the window length in calendar days (total_seconds / 86400), for backward compatibility with existing API consumers. The fixed/credit calculation independently counts whole local days as described above. All arithmetic uses Decimal; the returned dict contains Python floats for JSON-serialisation convenience. Parameters ---------- session: Active read-only SQLAlchemy session. start: Inclusive start of the summary interval (UTC or naive-UTC). end: Exclusive end of the summary interval (UTC or naive-UTC). Returns ------- dict with keys: currency str ISO 4217 currency (from contract, or "EUR" fallback) metered_import float Σ import_cost from non-degraded periods metered_export float Σ export_revenue from non-degraded periods metered_net float Σ net_cost from non-degraded periods fixed_costs float standing charges for elapsed whole local days credits float energy-tax credit for elapsed whole local days total_payable float metered_net + fixed_costs − credits period_count int number of non-degraded periods in range degraded_count int number of degraded periods in range days float interval length in days (total_seconds / 86400) """ from datetime import timedelta as _td, date as _date start_utc = _as_utc(start) end_utc = _as_utc(end) # --- Fetch all EnergyCostPeriod rows in [start, end) --- rows = session.execute( select(EnergyCostPeriod).where( EnergyCostPeriod.period_start >= start_utc, EnergyCostPeriod.period_start < end_utc, ) ).scalars().all() good_rows = [r for r in rows if not r.degraded] degraded_rows = [r for r in rows if r.degraded] # Σ monetary amounts (Decimal arithmetic). sum_import = sum((_to_decimal(r.import_cost) for r in good_rows), Decimal("0")) sum_export = sum((_to_decimal(r.export_revenue) for r in good_rows), Decimal("0")) sum_net = sum((_to_decimal(r.net_cost) for r in good_rows), Decimal("0")) # --- Interval length in days (window, not elapsed — kept for API compat) --- total_seconds = (end_utc - start_utc).total_seconds() days = _to_decimal(str(total_seconds)) / _to_decimal("86400") # --- Fixed costs and credits: Principle C cross-version whole-day counting --- today_local: _date = local_now().date() # --- Compute [first_counted, last_counted] local date range (inclusive) --- # # We count local calendar day D if its LOCAL MIDNIGHT falls within [start_utc, end_utc) # AND D ≤ today_local (only elapsed / today days count as "whole days"). # # Semantics: "A day D is counted when its local midnight has arrived (D ≤ today) # AND the local midnight is within the billing window." # # This means a sub-day window [10:00, 10:30) UTC that doesn't contain any # local midnight contributes 0 days, while a window [22:00 UTC, 23:00 UTC) that # contains CEST midnight (= 22:00 UTC) contributes 1 day. # # Implementation: compute the first and last local date whose midnight is in range. from app.services.timezone import local_midnight_utc as _lmu local_start_date: _date = local_date(start_utc) # Is the midnight of local_start_date ≥ start_utc? If not, first midnight is next day. if _lmu(local_start_date) >= start_utc: first_counted: _date = local_start_date else: first_counted = local_start_date + _td(days=1) local_end_date: _date = local_date(end_utc) # Is the midnight of local_end_date < end_utc? If yes, that day's midnight is in range. if _lmu(local_end_date) < end_utc: last_counted: _date = local_end_date else: last_counted = local_end_date - _td(days=1) # Cap: only elapsed or today's days. last_counted = min(last_counted, today_local) # If the range is empty (first_counted > last_counted), no days are counted. # Fetch all versions of the active contract once (single DB call). versions = active_contract_versions(session) fixed_dec = Decimal("0") credits_dec = Decimal("0") currency = "EUR" if versions: # Currency comes from the contract regardless of day count. currency = versions[0].contract.currency if versions and first_counted <= last_counted: # For each version segment, find its overlap with [first_counted, last_counted] # and accumulate whole days. version_segments: list[tuple[_date, _date | None, dict]] = [] for v in versions: v_start_local = local_date(_as_utc(v.effective_from)) v_end_local = local_date(_as_utc(v.effective_to)) if v.effective_to is not None else None version_segments.append((v_start_local, v_end_local, v.values or {})) for v_start, v_end_excl, v_values in version_segments: # The version covers [v_start, v_end_excl) in local dates, # where v_end_excl=None means open-ended (no upper bound). # Intersection with [first_counted, last_counted]: seg_start = max(first_counted, v_start) if v_end_excl is not None: seg_end_excl = min(last_counted + _td(days=1), v_end_excl) else: seg_end_excl = last_counted + _td(days=1) if seg_start >= seg_end_excl: continue # no overlap n_days = (seg_end_excl - seg_start).days if n_days <= 0: continue standing: dict = v_values.get("standing", {}) creds: dict = v_values.get("credits", {}) network_fee = _to_decimal(standing.get("network_fee", 0)) management_fee = _to_decimal(standing.get("management_fee", 0)) heffingskorting = _to_decimal(creds.get("heffingskorting", 0)) # Standing charges: EUR/month → EUR/day (÷ 30) × n_days. fixed_dec += (network_fee + management_fee) / Decimal("30") * Decimal(str(n_days)) # Energy-tax credit: EUR/year → EUR/day (÷ 365) × n_days. credits_dec += heffingskorting / Decimal("365") * Decimal(str(n_days)) total_payable = sum_net + fixed_dec - credits_dec return { "currency": currency, "metered_import": float(sum_import), "metered_export": float(sum_export), "metered_net": float(sum_net), "fixed_costs": float(fixed_dec), "credits": float(credits_dec), "total_payable": float(total_payable), "period_count": len(good_rows), "degraded_count": len(degraded_rows), "days": float(days), }