M7-T03: make billing engine meter-aware (no cross-meter delta, delta sanity guard, period.meter_id)
This commit is contained in:
+205
-27
@@ -1,7 +1,7 @@
|
||||
"""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:
|
||||
M6 design document, extended in M7-T03 to be meter-aware:
|
||||
|
||||
**Layer 1 — per-period metering cost (immutable, price-snapshot)**
|
||||
``compute_period(session, t0)`` computes the import cost, export revenue, and
|
||||
@@ -31,17 +31,49 @@ Design notes
|
||||
- **Register keys**: DSMR payload uses JSON strings like ``"20915.154"``
|
||||
for cumulative kWh registers. ``register_at`` converts them to Decimal.
|
||||
- **Degraded vs skip semantics**:
|
||||
- *No meter coverage* (``meter_at`` returns None for t0): write a
|
||||
``degraded=True`` row with ``meter_id=None``.
|
||||
- *Cross-meter boundary* (m0.id != m1.id for t0/t1): write a ``degraded=True``
|
||||
row with ``meter_id=m0.id``; losing this one period at the swap boundary is
|
||||
acceptable (D5 decision).
|
||||
- *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``.
|
||||
boundary within the meter window): write a ``degraded=True`` row with
|
||||
``meter_id=m0.id`` so the period is tracked and can be retried by
|
||||
``compute_closed_periods``.
|
||||
- *Negative or excessively large delta* (delta sanity guard D6): write a
|
||||
``degraded=True`` row with ``meter_id=m0.id``; prevents negative costs and
|
||||
grossly inflated costs from meter resets, DSMR rollover, or data spikes.
|
||||
- *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).
|
||||
- **Meter-aware register lookup**: ``register_at`` now accepts a ``meter``
|
||||
parameter and restricts the DSMR reading query to readings within
|
||||
``[meter.started_at, meter.ended_at)`` (half-open), preventing old-meter
|
||||
readings from leaking into a new-meter epoch.
|
||||
- **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.
|
||||
|
||||
Meter-aware compute_period ordering rationale (M7-T03)
|
||||
-------------------------------------------------------
|
||||
The order of checks inside ``compute_period`` is:
|
||||
|
||||
1. **Immutability guard** (existing non-degraded row, overwrite=False) → return False.
|
||||
2. **Meter determination** (m0 = meter_at(t0), m1 = meter_at(t1)):
|
||||
- No meter (m0 is None) → write degraded, meter_id=None.
|
||||
- Cross-meter boundary (m0.id != m1.id) → write degraded, meter_id=m0.id.
|
||||
3. **Active contract version check** → skip (no write) if absent.
|
||||
4. **Boundary register readings** within m0's window → write degraded if missing.
|
||||
5. **Delta sanity guard** → write degraded if any delta < 0 or > _MAX_DELTA_KWH.
|
||||
6. **Price strategy** → skip (no write) if Tibber price missing.
|
||||
7. **Upsert billing record** with meter_id=m0.id.
|
||||
|
||||
Why meter before contract? The meter is a *structural* prerequisite: without a
|
||||
known meter epoch we cannot trust the delta at all, so we commit a degraded row
|
||||
immediately. The contract skip, by contrast, is transient (the period can be
|
||||
re-computed once a contract is configured), so it produces no row.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -59,8 +91,9 @@ from app.integrations.pricing.strategies import (
|
||||
TibberPriceNotFoundError,
|
||||
get_strategy,
|
||||
)
|
||||
from app.models.energy import DsmrReading, EnergyCostPeriod
|
||||
from app.models.energy import DsmrReading, EnergyCostPeriod, Meter
|
||||
from app.services.contracts import active_contract_version_at, active_contract_versions
|
||||
from app.services.meters import meter_at
|
||||
from app.services.timezone import local_date, local_now
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -81,6 +114,15 @@ _LOOKBACK_DAYS = 7 # maximum lookback window for compute_closed_periods
|
||||
# producing a spurious zero-delta "successful" row.
|
||||
_READING_MAX_STALENESS = timedelta(minutes=_PERIOD_MINUTES)
|
||||
|
||||
# Maximum plausible kWh delta for a single 15-minute period (D6 sanity guard).
|
||||
# A typical Dutch household uses well under 5 kWh per quarter hour even under
|
||||
# heavy load. 100 kWh per 15 minutes corresponds to ~400 kW — far beyond any
|
||||
# residential consumption — but is lenient enough to never fire on legitimate
|
||||
# data. Any delta at or above this threshold indicates a meter reset, DSMR
|
||||
# rollover, sign error, or other data anomaly, and the period is marked
|
||||
# degraded to prevent negative costs or grossly inflated charges.
|
||||
_MAX_DELTA_KWH = Decimal("100")
|
||||
|
||||
# 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)
|
||||
@@ -123,15 +165,29 @@ def _existing_period(session: Session, t0: datetime) -> EnergyCostPeriod | None:
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# register_at — boundary reading lookup
|
||||
# register_at — boundary reading lookup (meter-aware)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def register_at(session: Session, boundary: datetime) -> dict[str, Decimal] | None:
|
||||
"""Return the four cumulative kWh register values at *boundary*.
|
||||
def register_at(
|
||||
session: Session,
|
||||
boundary: datetime,
|
||||
meter: Meter,
|
||||
) -> dict[str, Decimal] | None:
|
||||
"""Return the four cumulative kWh register values at *boundary*, within *meter*'s window.
|
||||
|
||||
Queries the most recent ``DsmrReading`` with ``recorded_at ≤ boundary``
|
||||
and extracts the four energy registers from ``payload``:
|
||||
Queries the most recent ``DsmrReading`` with:
|
||||
``recorded_at ≤ boundary``
|
||||
AND ``recorded_at ≥ meter.started_at``
|
||||
AND (``meter.ended_at IS NULL`` OR ``recorded_at < meter.ended_at``)
|
||||
|
||||
The meter window constraint (half-open ``[started_at, ended_at)``) ensures
|
||||
that readings from a previous meter epoch are never used to anchor a new
|
||||
meter's computation. Without this guard, the final reading of the old meter
|
||||
would be visible at the start of the new meter's epoch and produce a
|
||||
cross-meter delta, defeating the isolation guarantee.
|
||||
|
||||
Extracts the four energy registers from ``payload``:
|
||||
|
||||
d1 — electricity_delivered_1 (delivered low-tariff / dal)
|
||||
d2 — electricity_delivered_2 (delivered high-tariff / normal)
|
||||
@@ -142,18 +198,40 @@ def register_at(session: Session, boundary: datetime) -> dict[str, Decimal] | No
|
||||
-------
|
||||
dict[str, Decimal] with keys ``d1``, ``d2``, ``r1``, ``r2``, or ``None``
|
||||
when:
|
||||
- No ``DsmrReading`` row exists with ``recorded_at ≤ boundary``.
|
||||
- No ``DsmrReading`` row exists with ``recorded_at ≤ boundary`` within
|
||||
*meter*'s epoch window.
|
||||
- The most recent such reading is older than ``_READING_MAX_STALENESS``
|
||||
relative to *boundary* (freshness guard).
|
||||
- Any of the four register keys is absent from the payload.
|
||||
- Any of the four register values is ``None`` (null in JSON).
|
||||
|
||||
SQLite naive datetime note
|
||||
--------------------------
|
||||
``recorded_at`` is stored as a naive UTC datetime in SQLite. Comparisons
|
||||
against *boundary* (always tz-aware UTC) use ``_as_utc()`` for the
|
||||
freshness check. The SQL ``WHERE`` clause comparisons work correctly
|
||||
because SQLAlchemy's SQLite dialect strips tzinfo when binding parameters
|
||||
(leaving the wall-clock UTC value unchanged), consistent with the storage
|
||||
format.
|
||||
"""
|
||||
row: DsmrReading | None = (
|
||||
session.execute(
|
||||
select(DsmrReading)
|
||||
.where(DsmrReading.recorded_at <= boundary)
|
||||
.order_by(DsmrReading.recorded_at.desc())
|
||||
.limit(1)
|
||||
).scalar_one_or_none()
|
||||
# Build the meter-window constraints: [started_at, ended_at).
|
||||
meter_lower = meter.started_at # DsmrReading.recorded_at >= meter.started_at
|
||||
meter_upper = meter.ended_at # DsmrReading.recorded_at < meter.ended_at (if set)
|
||||
|
||||
stmt = (
|
||||
select(DsmrReading)
|
||||
.where(
|
||||
DsmrReading.recorded_at <= boundary,
|
||||
DsmrReading.recorded_at >= meter_lower,
|
||||
)
|
||||
.order_by(DsmrReading.recorded_at.desc())
|
||||
.limit(1)
|
||||
)
|
||||
# Apply the upper bound only when the meter is closed (ended_at is not None).
|
||||
if meter_upper is not None:
|
||||
stmt = stmt.where(DsmrReading.recorded_at < meter_upper)
|
||||
|
||||
row: DsmrReading | None = session.execute(stmt).scalar_one_or_none()
|
||||
|
||||
if row is None:
|
||||
return None
|
||||
@@ -214,8 +292,14 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
|
||||
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 no meter covers t0 (``meter_at`` returns None for t0): inserts/updates
|
||||
a degraded row with ``meter_id=None``.
|
||||
- If the period spans a meter boundary (``meter_at(t0).id != meter_at(t1).id``):
|
||||
inserts/updates a degraded row with ``meter_id=m0.id`` (D5 decision).
|
||||
- If readings are missing at either boundary within the meter window:
|
||||
inserts/updates a degraded row with ``meter_id=m0.id``.
|
||||
- If any delta is negative or exceeds ``_MAX_DELTA_KWH`` (D6 sanity guard):
|
||||
inserts/updates a degraded row with ``meter_id=m0.id``.
|
||||
- If the active contract version is missing: **skips** (returns False, no write).
|
||||
- If the Tibber price is missing (TibberPriceNotFoundError): **skips**
|
||||
(returns False, no write).
|
||||
@@ -229,7 +313,50 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
|
||||
if existing is not None and not existing.degraded and not overwrite:
|
||||
return False
|
||||
|
||||
# --- Active contract version at t0 (checked before readings) ---
|
||||
# --- Meter determination (structural prerequisite, checked before contract) ---
|
||||
#
|
||||
# A missing or cross-boundary meter is a structural problem: we cannot trust
|
||||
# the delta at all, so we write a degraded row immediately. This is different
|
||||
# from the contract skip (transient, no write): the degraded row ensures the
|
||||
# period appears in the history and can be revisited once the meter timeline
|
||||
# is corrected and a recompute_range is triggered.
|
||||
#
|
||||
# Ordering rationale:
|
||||
# 1. No meter (m0 is None) → degraded(meter_id=None): no epoch for t0.
|
||||
# 2. Cross-meter boundary (m0.id != m1.id) → degraded(meter_id=m0.id): D5.
|
||||
# 3. (Single meter, proceed) → contract check → readings → delta guard → price.
|
||||
#
|
||||
# We place meter before contract so that "cross-table period" is always
|
||||
# marked degraded regardless of contract state. If we checked contract
|
||||
# first, a missing-contract skip would silently discard the cross-table
|
||||
# evidence; once a contract is added and recompute runs, the engine would
|
||||
# incorrectly use cross-table reads.
|
||||
m0 = meter_at(session, t0)
|
||||
m1 = meter_at(session, t1)
|
||||
|
||||
if m0 is None:
|
||||
# No meter epoch covers t0 — degraded with no meter attribution.
|
||||
logger.debug(
|
||||
"compute_period(%s): no active meter at t0 — writing degraded (meter_id=None).",
|
||||
t0.isoformat(),
|
||||
)
|
||||
_upsert_degraded(session, t0, now, existing, meter_id=None)
|
||||
return True
|
||||
|
||||
if m1 is None or m0.id != m1.id:
|
||||
# Period spans a meter boundary or t1 has no meter. Degrade with m0's id
|
||||
# (t0's meter attribution): the period's start belongs to m0's epoch.
|
||||
logger.debug(
|
||||
"compute_period(%s): period crosses meter boundary "
|
||||
"(m0.id=%s, m1.id=%s) — writing degraded.",
|
||||
t0.isoformat(),
|
||||
m0.id,
|
||||
m1.id if m1 is not None else None,
|
||||
)
|
||||
_upsert_degraded(session, t0, now, existing, meter_id=m0.id)
|
||||
return True
|
||||
|
||||
# --- Active contract version at t0 ---
|
||||
# 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
|
||||
@@ -240,14 +367,13 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
|
||||
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)
|
||||
# --- Boundary readings within m0's meter window ---
|
||||
start_regs = register_at(session, t0, m0)
|
||||
end_regs = register_at(session, t1, m0)
|
||||
|
||||
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)
|
||||
# Missing readings within the meter window → degraded with m0 attribution.
|
||||
_upsert_degraded(session, t0, now, existing, meter_id=m0.id)
|
||||
return True # a record was written (degraded)
|
||||
|
||||
# --- Compute deltas (end − start) ---
|
||||
@@ -258,6 +384,26 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
|
||||
r2=end_regs["r2"] - start_regs["r2"],
|
||||
)
|
||||
|
||||
# --- Delta sanity guard (D6) ---
|
||||
# Any negative delta indicates a meter reset, DSMR rollover, or data error.
|
||||
# Any delta exceeding _MAX_DELTA_KWH (100 kWh per 15 min = 400 kW average)
|
||||
# is implausible for residential use and indicates an anomaly.
|
||||
# Both cases produce a degraded row so no negative or grossly inflated cost
|
||||
# is ever written to the billing record.
|
||||
all_deltas = (deltas.d1, deltas.d2, deltas.r1, deltas.r2)
|
||||
if any(d < Decimal("0") for d in all_deltas) or any(d > _MAX_DELTA_KWH for d in all_deltas):
|
||||
logger.debug(
|
||||
"compute_period(%s): delta sanity guard triggered "
|
||||
"(d1=%s, d2=%s, r1=%s, r2=%s) — writing degraded.",
|
||||
t0.isoformat(),
|
||||
deltas.d1,
|
||||
deltas.d2,
|
||||
deltas.r1,
|
||||
deltas.r2,
|
||||
)
|
||||
_upsert_degraded(session, t0, now, existing, meter_id=m0.id)
|
||||
return True
|
||||
|
||||
# --- Price strategy ---
|
||||
strategy = get_strategy(version.contract.kind)
|
||||
try:
|
||||
@@ -288,6 +434,7 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
|
||||
existing.currency = version.contract.currency
|
||||
existing.pricing = pricing
|
||||
existing.contract_version_id = version.id
|
||||
existing.meter_id = m0.id
|
||||
existing.degraded = False
|
||||
existing.computed_at = now
|
||||
else:
|
||||
@@ -303,6 +450,7 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
|
||||
currency=version.contract.currency,
|
||||
pricing=pricing,
|
||||
contract_version_id=version.id,
|
||||
meter_id=m0.id,
|
||||
degraded=False,
|
||||
computed_at=now,
|
||||
)
|
||||
@@ -316,9 +464,25 @@ def _upsert_degraded(
|
||||
t0: datetime,
|
||||
now: datetime,
|
||||
existing: EnergyCostPeriod | None,
|
||||
*,
|
||||
meter_id: int | None,
|
||||
) -> None:
|
||||
"""Insert or update a degraded placeholder for period *t0*.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
session:
|
||||
Active SQLAlchemy session.
|
||||
t0:
|
||||
UTC start of the 15-minute period.
|
||||
now:
|
||||
Current UTC timestamp for the ``computed_at`` field.
|
||||
existing:
|
||||
The existing ``EnergyCostPeriod`` row for this period, or ``None``.
|
||||
meter_id:
|
||||
The meter ID to attribute this degraded period to, or ``None`` when
|
||||
no meter epoch covers the period (no-meter degraded case).
|
||||
|
||||
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
|
||||
@@ -326,6 +490,11 @@ def _upsert_degraded(
|
||||
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.
|
||||
|
||||
The ``meter_id`` is always updated to reflect the current meter attribution
|
||||
judgment (the result of ``meter_at`` at the time of recompute). This
|
||||
ensures that a retroactive ``started_at`` change + ``recompute_range`` will
|
||||
re-attribute historical degraded periods to the correct meter epoch.
|
||||
"""
|
||||
if existing is not None:
|
||||
# Explicitly reset to degraded state — identical field values to the
|
||||
@@ -340,6 +509,7 @@ def _upsert_degraded(
|
||||
existing.net_cost = 0.0
|
||||
existing.pricing = {}
|
||||
existing.contract_version_id = None
|
||||
existing.meter_id = meter_id
|
||||
existing.degraded = True
|
||||
existing.computed_at = now
|
||||
else:
|
||||
@@ -355,6 +525,7 @@ def _upsert_degraded(
|
||||
currency="EUR", # placeholder; real currency known after contract lookup
|
||||
pricing={},
|
||||
contract_version_id=None,
|
||||
meter_id=meter_id,
|
||||
degraded=True,
|
||||
computed_at=now,
|
||||
)
|
||||
@@ -381,6 +552,7 @@ def compute_closed_periods(session: Session) -> int:
|
||||
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.
|
||||
- Writes degraded rows for periods with no meter or cross-meter boundaries.
|
||||
- Skips periods for which no active contract version exists or the
|
||||
Tibber price is unavailable (without writing a degraded row).
|
||||
|
||||
@@ -429,11 +601,17 @@ def recompute_range(session: Session, start: datetime, end: datetime) -> int:
|
||||
This is the *explicit opt-in* path for recovering from:
|
||||
- Periods where readings or prices arrived late.
|
||||
- Price corrections (new contract version retroactively applied).
|
||||
- Retroactive meter changes (``update_meter`` with new ``started_at``) —
|
||||
re-running this function will re-judge meter attribution and re-compute
|
||||
costs using the corrected epoch boundaries.
|
||||
- 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.
|
||||
Existing rows (including successful ones) are overwritten; their
|
||||
``meter_id`` fields will reflect the *current* ``meter_at`` judgment for
|
||||
each period's start timestamp, naturally re-attributing periods when meter
|
||||
``started_at`` values have been retroactively corrected.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
|
||||
Reference in New Issue
Block a user