M8-T14: add generic meter cost periods

alembic_app/versions/20260822_19_meter_cost_periods.py

@@ -0,0 +1,95 @@
+"""add generic commodity-scoped meter cost periods
+
+Revision ID: 20260822_19_meter_cost_periods
+Revises: 20260822_18_contract_scopes
+Create Date: 2026-08-22 00:00:00.000000
+
+This additive migration creates a separate audit ledger for non-electricity
+meter costs. It deliberately does not alter, migrate, or delete rows from the
+existing electricity-only energy_cost_period table.
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision: str = "20260822_19_meter_cost_periods"
+down_revision: Union[str, None] = "20260822_18_contract_scopes"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+class ExactDecimal(sa.TypeDecorator):
+    """Use SQLite text storage while retaining Numeric semantics elsewhere."""
+
+    impl = sa.Numeric
+    cache_ok = True
+
+    def __init__(self, precision: int, scale: int) -> None:
+        self.precision = precision
+        self.scale = scale
+        super().__init__(precision=precision, scale=scale)
+
+    def load_dialect_impl(self, dialect):
+        if dialect.name == "sqlite":
+            return dialect.type_descriptor(sa.String(self.precision + 2))
+        return dialect.type_descriptor(sa.Numeric(self.precision, self.scale, asdecimal=True))
+
+
+def upgrade() -> None:
+    op.create_table(
+        "meter_cost_period",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("commodity", sa.String(length=32), nullable=False),
+        sa.Column("period_start", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("period_end", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("meter_id", sa.Integer(), nullable=True),
+        sa.Column("source_binding_id", sa.Integer(), nullable=True),
+        sa.Column("contract_version_id", sa.Integer(), nullable=True),
+        # SQLite NUMERIC coercion binds Decimal values as binary floats.  Store
+        # fixed-width decimal text there, while retaining Numeric semantics on
+        # other supported dialects.
+        sa.Column("quantity", ExactDecimal(15, 6), nullable=False),
+        sa.Column("cost", ExactDecimal(15, 9), nullable=False),
+        sa.Column("currency", sa.String(length=8), nullable=False),
+        sa.Column("cost_breakdown", sa.JSON(), nullable=False),
+        sa.Column("pricing_snapshot", sa.JSON(), nullable=False),
+        sa.Column("quality", sa.String(length=32), nullable=False),
+        sa.Column("degraded", sa.Boolean(), nullable=False, server_default=sa.false()),
+        sa.Column("degraded_reason", sa.String(length=255), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["meter_id"], ["meter.id"], ondelete="RESTRICT"),
+        sa.ForeignKeyConstraint(
+            ["source_binding_id"], ["meter_source_binding.id"], ondelete="RESTRICT"
+        ),
+        sa.ForeignKeyConstraint(
+            ["contract_version_id"], ["energy_contract_version.id"], ondelete="RESTRICT"
+        ),
+        sa.CheckConstraint(
+            "degraded OR (meter_id IS NOT NULL AND source_binding_id IS NOT NULL "
+            "AND contract_version_id IS NOT NULL)",
+            name="ck_meter_cost_period_normal_audit_links",
+        ),
+        sa.CheckConstraint("period_end > period_start", name="ck_meter_cost_period_positive_interval"),
+        sa.CheckConstraint(
+            "NOT degraded OR (degraded_reason IS NOT NULL AND length(trim(degraded_reason)) > 0)",
+            name="ck_meter_cost_period_degraded_reason",
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("commodity", "period_start", name="uq_meter_cost_period_commodity_start"),
+    )
+    op.create_index(
+        "ix_meter_cost_period_commodity_start", "meter_cost_period", ["commodity", "period_start"]
+    )
+    op.create_index(
+        "ix_meter_cost_period_source_binding_id", "meter_cost_period", ["source_binding_id"]
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_meter_cost_period_source_binding_id", table_name="meter_cost_period")
+    op.drop_index("ix_meter_cost_period_commodity_start", table_name="meter_cost_period")
+    op.drop_table("meter_cost_period")

app/models/energy.py

@@ -12,20 +12,25 @@ Six tables:
 from __future__ import annotations
 
 import uuid as _uuid
-from datetime import datetime
+from datetime import datetime, timezone
+from decimal import Decimal
+from typing import Any
 from sqlalchemy import (
     Boolean,
+    CheckConstraint,
     DateTime,
     Float,
     ForeignKey,
+    Index,
     Integer,
+    Numeric,
     String,
     UniqueConstraint,
     event,
     text,
 )
-from sqlalchemy.orm import Mapped, mapped_column, relationship, synonym
-from sqlalchemy.types import JSON
+from sqlalchemy.orm import Mapped, mapped_column, relationship, synonym, validates
+from sqlalchemy.types import JSON, TypeDecorator
 
 from app.db import Base
 from app.models.meter_source import MeterSourceBinding
@@ -35,6 +40,106 @@ def _uuid4_str() -> str:
     return str(_uuid.uuid4())
 
 
+def _decimal_json(value: Any) -> Any:
+    """Make auditable JSON portable without admitting binary numeric values."""
+    if isinstance(value, Decimal):
+        return format(value, "f")
+    if isinstance(value, float) or (isinstance(value, int) and not isinstance(value, bool)):
+        raise ValueError("JSON amounts and quantities must be decimal strings, not numeric JSON values")
+    if isinstance(value, dict):
+        return {key: _decimal_json(child) for key, child in value.items()}
+    if isinstance(value, list):
+        return [_decimal_json(child) for child in value]
+    return value
+
+
+def _validate_fixed_decimal(value: Decimal, precision: int, scale: int, field: str) -> Decimal:
+    if not isinstance(value, Decimal):
+        raise ValueError(f"{field} must be a Decimal, not a binary float or other numeric type")
+    if not value.is_finite():
+        raise ValueError(f"{field} must be finite")
+    if -value.as_tuple().exponent > scale:
+        raise ValueError(f"{field} exceeds scale {scale}")
+    integer_digits = 0 if value.is_zero() else max(value.copy_abs().adjusted() + 1, 0)
+    if integer_digits > precision - scale:
+        raise ValueError(f"{field} exceeds precision {precision},{scale}")
+    return value
+
+
+class ExactDecimal(TypeDecorator[Decimal]):
+    """Fixed-point Decimal which uses SQLite TEXT, never a binary float."""
+
+    impl = Numeric
+    cache_ok = True
+
+    def __init__(self, precision: int, scale: int) -> None:
+        self.precision = precision
+        self.scale = scale
+        super().__init__(precision=precision, scale=scale)
+
+    def load_dialect_impl(self, dialect):
+        if dialect.name == "sqlite":
+            return dialect.type_descriptor(String(self.precision + 2))
+        return dialect.type_descriptor(Numeric(self.precision, self.scale, asdecimal=True))
+
+    def process_bind_param(self, value: Decimal | None, dialect) -> Decimal | str | None:
+        if value is None:
+            return None
+        value = _validate_fixed_decimal(value, self.precision, self.scale, "decimal value")
+        if dialect.name == "sqlite":
+            return format(value, f".{self.scale}f")
+        return value
+
+    def process_result_value(self, value: Decimal | str | None, _dialect) -> Decimal | None:
+        return None if value is None else Decimal(value)
+
+
+class DecimalJSON(TypeDecorator[dict]):
+    """JSON which serializes Decimal values as strings on every write path."""
+
+    impl = JSON
+    cache_ok = True
+
+    def process_bind_param(self, value: Any, _dialect) -> Any:
+        return None if value is None else _decimal_json(value)
+
+
+class UTCDateTime(TypeDecorator[datetime]):
+    """UTC timestamps that preserve instant identity on SQLite and other dialects."""
+
+    impl = DateTime(timezone=True)
+    cache_ok = True
+
+    def __init__(self, field: str) -> None:
+        self.field = field
+        super().__init__()
+
+    def process_bind_param(self, value: datetime | None, _dialect) -> datetime | None:
+        return None if value is None else _normalise_utc_period(value, self.field)
+
+    def process_result_value(self, value: datetime | None, _dialect) -> datetime | None:
+        if value is None:
+            return None
+        if value.tzinfo is None or value.utcoffset() is None:
+            return value.replace(tzinfo=timezone.utc)
+        return value.astimezone(timezone.utc)
+
+
+def _require_aware_period(start: datetime, end: datetime) -> None:
+    if start.tzinfo is None or start.utcoffset() is None:
+        raise ValueError("period_start must be timezone-aware")
+    if end.tzinfo is None or end.utcoffset() is None:
+        raise ValueError("period_end must be timezone-aware")
+    if end <= start:
+        raise ValueError("period_end must be after period_start")
+
+
+def _normalise_utc_period(value: datetime, field: str) -> datetime:
+    if value.tzinfo is None or value.utcoffset() is None:
+        raise ValueError(f"{field} must be timezone-aware")
+    return value.astimezone(timezone.utc)
+
+
 class Meter(Base):
     """One physical electricity meter's installation epoch.
 
@@ -358,6 +463,91 @@ class EnergyCostPeriod(Base):
     )
 
 
+class MeterCostPeriod(Base):
+    """Auditable commodity-scoped ledger row for one half-open metering period.
+
+    A degraded row intentionally permits missing audit links; services must
+    still require them before writing a normal row.
+    """
+
+    __tablename__ = "meter_cost_period"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    commodity: Mapped[str] = mapped_column(String(32), nullable=False)
+    period_start: Mapped[datetime] = mapped_column(UTCDateTime("period_start"), nullable=False)
+    period_end: Mapped[datetime] = mapped_column(UTCDateTime("period_end"), nullable=False)
+    meter_id: Mapped[int | None] = mapped_column(
+        ForeignKey("meter.id", ondelete="RESTRICT"), nullable=True
+    )
+    source_binding_id: Mapped[int | None] = mapped_column(
+        ForeignKey("meter_source_binding.id", ondelete="RESTRICT"), nullable=True
+    )
+    contract_version_id: Mapped[int | None] = mapped_column(
+        ForeignKey("energy_contract_version.id", ondelete="RESTRICT"), nullable=True
+    )
+
+    # SQLite reliably round-trips at most fifteen significant decimal digits.
+    # WarmteLink itself reports 0.001 units, so nine cost fractional digits
+    # retain a six-place tariff times that source precision without float loss.
+    quantity: Mapped[Decimal] = mapped_column(ExactDecimal(15, 6), nullable=False)
+    cost: Mapped[Decimal] = mapped_column(ExactDecimal(15, 9), nullable=False)
+    currency: Mapped[str] = mapped_column(String(8), nullable=False)
+    cost_breakdown: Mapped[dict] = mapped_column(DecimalJSON(), nullable=False, default=dict)
+    pricing_snapshot: Mapped[dict] = mapped_column(DecimalJSON(), nullable=False, default=dict)
+    quality: Mapped[str] = mapped_column(String(32), nullable=False, default="valid")
+    degraded: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    degraded_reason: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+
+    meter: Mapped["Meter | None"] = relationship()
+    source_binding: Mapped["MeterSourceBinding | None"] = relationship()
+    contract_version: Mapped["EnergyContractVersion | None"] = relationship()
+
+    __table_args__ = (
+        CheckConstraint(
+            "degraded OR (meter_id IS NOT NULL AND source_binding_id IS NOT NULL "
+            "AND contract_version_id IS NOT NULL)",
+            name="ck_meter_cost_period_normal_audit_links",
+        ),
+        CheckConstraint("period_end > period_start", name="ck_meter_cost_period_positive_interval"),
+        CheckConstraint(
+            "NOT degraded OR (degraded_reason IS NOT NULL AND length(trim(degraded_reason)) > 0)",
+            name="ck_meter_cost_period_degraded_reason",
+        ),
+        UniqueConstraint("commodity", "period_start", name="uq_meter_cost_period_commodity_start"),
+        Index("ix_meter_cost_period_commodity_start", "commodity", "period_start"),
+        Index("ix_meter_cost_period_source_binding_id", "source_binding_id"),
+    )
+
+    @validates("cost_breakdown", "pricing_snapshot")
+    def _validate_decimal_json(self, _key: str, value: dict) -> dict:
+        return _decimal_json(value)
+
+    @validates("quantity", "cost")
+    def _validate_fixed_decimal(self, key: str, value: Decimal) -> Decimal:
+        precision, scale = (15, 6) if key == "quantity" else (15, 9)
+        return _validate_fixed_decimal(value, precision, scale, key)
+
+    @validates("period_start", "period_end")
+    def _normalise_period(self, key: str, value: datetime) -> datetime:
+        return _normalise_utc_period(value, key)
+
+
+@event.listens_for(MeterCostPeriod, "before_insert")
+@event.listens_for(MeterCostPeriod, "before_update")
+def _validate_meter_cost_period(_mapper, _connection, target: MeterCostPeriod) -> None:
+    _require_aware_period(target.period_start, target.period_end)
+    if not target.degraded and (
+        target.meter_id is None
+        or target.source_binding_id is None
+        or target.contract_version_id is None
+    ):
+        raise ValueError("normal meter cost periods require meter, binding, and contract version")
+    if target.degraded and not target.degraded_reason:
+        raise ValueError("degraded meter cost periods require a degraded_reason")
+
+
 # Index on recorded_at for efficient time-range queries on DSMR readings.
 # (The ORM-level index=True on recorded_at already creates ix_dsmr_reading_recorded_at;
 # no composite index is needed for single-meter deployments.)

docs/design/m8-warmtelink-energy.md

@@ -810,7 +810,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接
 
 ### M8-T14 — 建立通用 Meter Cost Period 账本
 
-- **Status**: `todo`
+- **Status**: `done`
 - **Depends**: M8-T13
 - **Context**: 热力成本不能硬塞进 electricity 专用账本；先建立按 commodity 审计的独立表。
 
@@ -819,6 +819,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接
 - `create  alembic_app/versions/20260822_19_meter_cost_periods.py`
 - `modify  scripts/app_db_adopt.py`
 - `create  tests/test_meter_cost_models.py`
+- `modify  tests/test_energy_models.py`
 
 **Steps**
 1. 新建 `MeterCostPeriod`，字段按 §5：commodity、period start/end、nullable Meter/binding、nullable

scripts/app_db_adopt.py

@@ -15,7 +15,7 @@ if str(PROJECT_ROOT) not in sys.path:
 
 from app.config import get_settings
 
-APP_BASELINE_REVISION = "20260822_18_contract_scopes"
+APP_BASELINE_REVISION = "20260822_19_meter_cost_periods"
 
 
 class AppDatabaseAdoptionError(RuntimeError):

tests/test_energy_models.py

@@ -1251,8 +1251,8 @@ def test_contract_scope_migration_preserves_historical_contract_audit(tmp_path:
             {"now": now, "pricing": pricing, "version_id": version_id},
         )
 
-    command.upgrade(cfg, "head")
-    command.upgrade(cfg, "head")
+    command.upgrade(cfg, "20260822_18_contract_scopes")
+    command.upgrade(cfg, "20260822_18_contract_scopes")
     with engine.connect() as connection:
         assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == (
             "20260822_18_contract_scopes"
@@ -1359,7 +1359,7 @@ def test_contract_scope_migration_audit_failure_restores_revision_17(tmp_path: P
     engine.dispose()
 
     del cfg.attributes["m8_t12_post_ddl_audit_failure"]
-    command.upgrade(cfg, "head")
+    command.upgrade(cfg, "20260822_18_contract_scopes")
     engine = _engine_with_fk(db_url)
     with engine.connect() as connection:
         assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == (

tests/test_meter_cost_models.py

@@ -0,0 +1,465 @@
+"""Migration and ORM contracts for the commodity-scoped meter cost ledger."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, 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, event as sa_event, insert, inspect, select, text
+from sqlalchemy.exc import IntegrityError, StatementError
+from sqlalchemy.orm import Session
+
+from app.models.energy import MeterCostPeriod
+from scripts.app_db_adopt import APP_BASELINE_REVISION
+
+
+REVISION_18 = "20260822_18_contract_scopes"
+REVISION_19 = "20260822_19_meter_cost_periods"
+UTC = timezone.utc
+
+
+def _config(database_url: str) -> Config:
+    config = Config("alembic_app.ini")
+    config.set_main_option("sqlalchemy.url", database_url)
+    return config
+
+
+def _fk_engine(database_url: str):
+    engine = create_engine(database_url)
+
+    @sa_event.listens_for(engine, "connect")
+    def _enable_foreign_keys(connection, _record) -> None:
+        connection.execute("PRAGMA foreign_keys = ON")
+
+    return engine
+
+
+def _period(**overrides) -> MeterCostPeriod:
+    start = datetime(2026, 8, 22, 12, tzinfo=UTC)
+    values = {
+        "commodity": "heating",
+        "period_start": start,
+        "period_end": start + timedelta(minutes=15),
+        "quantity": Decimal("999999999.123456"),
+        "cost": Decimal("999999.123456789"),
+        "currency": "EUR",
+        "cost_breakdown": {"heating": "0.012345678"},
+        "pricing_snapshot": {"heating": "10.000000", "kind": "district_heating"},
+        "quality": "valid",
+        "degraded": True,
+        "degraded_reason": "test_fixture_without_audit_links",
+        "created_at": start,
+        "updated_at": start,
+    }
+    values.update(overrides)
+    return MeterCostPeriod(**values)
+
+
+def test_meter_cost_period_empty_db_upgrade_shape_and_baseline(tmp_path: Path) -> None:
+    database_url = f"sqlite:///{tmp_path / 'empty.db'}"
+    config = _config(database_url)
+    command.upgrade(config, "head")
+    command.upgrade(config, "head")
+
+    engine = create_engine(database_url)
+    inspector = inspect(engine)
+    columns = {column["name"]: column for column in inspector.get_columns("meter_cost_period")}
+    assert APP_BASELINE_REVISION == REVISION_19
+    assert {"commodity", "period_start", "period_end", "quantity", "cost"} <= columns.keys()
+    assert columns["meter_id"]["nullable"]
+    assert columns["source_binding_id"]["nullable"]
+    assert columns["contract_version_id"]["nullable"]
+    assert {index["name"] for index in inspector.get_indexes("meter_cost_period")} >= {
+        "ix_meter_cost_period_commodity_start",
+        "ix_meter_cost_period_source_binding_id",
+    }
+    assert ("commodity", "period_start") in {
+        tuple(constraint["column_names"])
+        for constraint in inspector.get_unique_constraints("meter_cost_period")
+    }
+    foreign_keys = {
+        foreign_key["constrained_columns"][0]: foreign_key
+        for foreign_key in inspector.get_foreign_keys("meter_cost_period")
+    }
+    for column, table in {
+        "meter_id": "meter",
+        "source_binding_id": "meter_source_binding",
+        "contract_version_id": "energy_contract_version",
+    }.items():
+        assert foreign_keys[column]["referred_table"] == table
+        assert foreign_keys[column]["options"]["ondelete"] == "RESTRICT"
+    engine.dispose()
+
+
+def test_meter_cost_period_revision_18_upgrade_preserves_electricity_rows(tmp_path: Path) -> None:
+    database_url = f"sqlite:///{tmp_path / 'revision18.db'}"
+    config = _config(database_url)
+    command.upgrade(config, REVISION_18)
+    engine = create_engine(database_url)
+    stamp = datetime(2026, 8, 22, 12)
+    with engine.begin() as connection:
+        connection.execute(
+            text(
+                "INSERT INTO energy_cost_period "
+                "(period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, import_cost, "
+                "export_revenue, net_cost, currency, pricing, contract_version_id, "
+                "meter_id, source_binding_id, degraded, computed_at) "
+                "VALUES (:stamp, 1, 2, 3, 4, 5, 6, 7, 'EUR', :pricing, NULL, NULL, NULL, 0, :stamp)"
+            ),
+            {"stamp": stamp, "pricing": '{"historic":"unchanged"}'},
+        )
+    command.upgrade(config, "head")
+    command.upgrade(config, "head")
+    with engine.connect() as connection:
+        assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == REVISION_19
+        assert connection.execute(text("SELECT COUNT(*) FROM energy_cost_period")).scalar_one() == 1
+        assert connection.execute(text("SELECT pricing FROM energy_cost_period")).scalar_one() == (
+            '{"historic":"unchanged"}'
+        )
+        assert connection.execute(text("SELECT COUNT(*) FROM meter_cost_period")).scalar_one() == 0
+    command.downgrade(config, REVISION_18)
+    assert "meter_cost_period" not in inspect(engine).get_table_names()
+    engine.dispose()
+
+
+def test_meter_cost_period_decimal_json_and_degraded_round_trip(tmp_path: Path) -> None:
+    database_url = f"sqlite:///{tmp_path / 'roundtrip.db'}"
+    command.upgrade(_config(database_url), "head")
+    engine = create_engine(database_url)
+    with Session(engine) as session:
+        normal = _period()
+        degraded = _period(
+            commodity="hot_water",
+            meter_id=None,
+            source_binding_id=None,
+            contract_version_id=None,
+            quality="unverifiable",
+            degraded=True,
+            degraded_reason="missing_binding",
+            quantity=Decimal("0.000000"),
+            cost=Decimal("0.000000000"),
+            cost_breakdown={"hot_water": "0.000000000"},
+            pricing_snapshot={"reason": "missing_binding"},
+        )
+        session.add_all((normal, degraded))
+        session.commit()
+        session.expire_all()
+        stored = session.get(MeterCostPeriod, normal.id)
+        assert stored is not None
+        assert stored.quantity == Decimal("999999999.123456")
+        assert stored.cost == Decimal("999999.123456789")
+        assert stored.cost_breakdown == {"heating": "0.012345678"}
+        assert all(not isinstance(value, float) for value in stored.pricing_snapshot.values())
+        assert session.get(MeterCostPeriod, degraded.id).degraded_reason == "missing_binding"
+    engine.dispose()
+
+
+def test_meter_cost_period_uses_decimal_text_and_normalises_json_snapshots(tmp_path: Path) -> None:
+    database_url = f"sqlite:///{tmp_path / 'exact.db'}"
+    command.upgrade(_config(database_url), "head")
+    engine = create_engine(database_url)
+    with Session(engine) as session:
+        row = _period(
+            cost_breakdown={"nested": [Decimal("2.000000000")]},
+            pricing_snapshot={"rate": Decimal("2.000000")},
+        )
+        session.add(row)
+        session.commit()
+        session.expire_all()
+        stored = session.get(MeterCostPeriod, row.id)
+        assert stored is not None
+        assert stored.cost_breakdown == {"nested": ["2.000000000"]}
+        assert stored.pricing_snapshot == {"rate": "2.000000"}
+        raw = session.execute(text("SELECT typeof(quantity), typeof(cost) FROM meter_cost_period")).one()
+        assert raw == ("text", "text")
+    with pytest.raises(ValueError, match="numeric JSON"):
+        _period(cost_breakdown={"nested": [1]})
+    with pytest.raises(ValueError, match="numeric JSON"):
+        _period(pricing_snapshot={"nested": [1.25]})
+    engine.dispose()
+
+
+def test_meter_cost_period_core_json_bind_normalises_decimal_and_rejects_numeric(tmp_path: Path) -> None:
+    database_url = f"sqlite:///{tmp_path / 'core-json.db'}"
+    command.upgrade(_config(database_url), "head")
+    engine = create_engine(database_url)
+    values = {
+        "commodity": "heating",
+        "period_start": datetime(2026, 8, 22, 12, tzinfo=UTC),
+        "period_end": datetime(2026, 8, 22, 12, 15, tzinfo=UTC),
+        "quantity": Decimal("0.000000"),
+        "cost": Decimal("0.000000000"),
+        "currency": "EUR",
+        "cost_breakdown": {"nested": [Decimal("2.000000000")]},
+        "pricing_snapshot": {"rate": Decimal("2.000000")},
+        "quality": "valid",
+        "degraded": True,
+        "degraded_reason": "core fixture",
+        "created_at": datetime(2026, 8, 22, 12, tzinfo=UTC),
+        "updated_at": datetime(2026, 8, 22, 12, tzinfo=UTC),
+    }
+    with engine.begin() as connection:
+        connection.execute(insert(MeterCostPeriod.__table__).values(values))
+    with Session(engine) as session:
+        stored = session.scalar(select(MeterCostPeriod))
+        assert stored is not None
+        assert stored.cost_breakdown == {"nested": ["2.000000000"]}
+        assert stored.pricing_snapshot == {"rate": "2.000000"}
+    for field, numeric_value in (("cost_breakdown", {"nested": [1]}), ("pricing_snapshot", {"rate": 1.25})):
+        with engine.begin() as connection, pytest.raises(StatementError, match="numeric JSON"):
+            connection.execute(
+                insert(MeterCostPeriod.__table__).values(
+                    {**values, "commodity": f"invalid-{field}", field: numeric_value}
+                )
+            )
+    engine.dispose()
+
+
+@pytest.mark.parametrize(
+    ("field", "value", "message"),
+    [
+        ("quantity", Decimal("1000000000.000000"), "precision"),
+        ("quantity", Decimal("1.1234567"), "scale"),
+        ("cost", Decimal("1000000.000000000"), "precision"),
+        ("cost", Decimal("1.1234567899"), "scale"),
+    ],
+)
+def test_meter_cost_period_rejects_decimal_precision_and_scale_overflow(field, value, message) -> None:
+    with pytest.raises(ValueError, match=message):
+        _period(**{field: value})
+
+
+def test_meter_cost_period_accepts_signed_decimal_boundaries_without_float_bind(tmp_path: Path) -> None:
+    database_url = f"sqlite:///{tmp_path / 'boundaries.db'}"
+    command.upgrade(_config(database_url), "head")
+    engine = create_engine(database_url)
+    with Session(engine) as session:
+        rows = (
+            _period(quantity=Decimal("999999999.999999"), cost=Decimal("999999.999999999")),
+            _period(
+                commodity="hot_water",
+                quantity=Decimal("-999999999.999999"),
+                cost=Decimal("-999999.999999999"),
+            ),
+        )
+        session.add_all(rows)
+        session.commit()
+        session.expire_all()
+        assert session.get(MeterCostPeriod, rows[0].id).quantity == Decimal("999999999.999999")
+        assert session.get(MeterCostPeriod, rows[1].id).cost == Decimal("-999999.999999999")
+    engine.dispose()
+
+
+def test_meter_cost_period_rejects_duplicate_period_invalid_interval_and_json_float(tmp_path: Path) -> None:
+    database_url = f"sqlite:///{tmp_path / 'constraints.db'}"
+    command.upgrade(_config(database_url), "head")
+    engine = create_engine(database_url)
+    with Session(engine) as session:
+        first = _period()
+        session.add(first)
+        session.commit()
+        session.add(_period(cost=Decimal("1.000000000")))
+        with pytest.raises(IntegrityError):
+            session.commit()
+        session.rollback()
+        session.add(_period(commodity="hot_water", period_end=datetime(2026, 8, 22, 12, tzinfo=UTC)))
+        with pytest.raises(ValueError, match="after period_start"):
+            session.flush()
+        session.rollback()
+        with pytest.raises(ValueError, match="timezone-aware"):
+            _period(
+                commodity="hot_water",
+                period_start=datetime(2026, 8, 22, 12),
+                period_end=datetime(2026, 8, 22, 12, 15, tzinfo=UTC),
+            )
+        with pytest.raises(ValueError, match="decimal strings"):
+            _period(pricing_snapshot={"heating": 1.25})
+        with pytest.raises(ValueError, match="normal meter cost periods"):
+            session.add(_period(commodity="hot_water", degraded=False, degraded_reason=None))
+            session.flush()
+        session.rollback()
+    engine.dispose()
+
+
+def test_meter_cost_period_uses_utc_instant_idempotency_and_db_constraints(tmp_path: Path) -> None:
+    database_url = f"sqlite:///{tmp_path / 'utc.db'}"
+    command.upgrade(_config(database_url), "head")
+    engine = create_engine(database_url)
+    utc_start = datetime(2026, 8, 22, 12, tzinfo=UTC)
+    offset_start = datetime(2026, 8, 22, 14, tzinfo=timezone(timedelta(hours=2)))
+    with Session(engine) as session:
+        first = _period(period_start=utc_start, period_end=utc_start + timedelta(minutes=15))
+        assert first.period_start == utc_start
+        session.add(first)
+        session.commit()
+        session.add(
+            _period(
+                period_start=offset_start,
+                period_end=offset_start + timedelta(minutes=15),
+                cost=Decimal("1.000000000"),
+            )
+        )
+        with pytest.raises(IntegrityError):
+            session.commit()
+        session.rollback()
+    with engine.begin() as connection:
+        values = {
+            "start": "2026-08-23 00:00:00.000000",
+            "quantity": "0.000000",
+            "cost": "0.000000000",
+            "now": "2026-08-23 00:00:00.000000",
+        }
+        for commodity, end in (("equal", values["start"]), ("reversed", "2026-08-22 23:59:59.000000")):
+            with pytest.raises(IntegrityError):
+                connection.execute(
+                    text(
+                        "INSERT INTO meter_cost_period "
+                        "(commodity, period_start, period_end, quantity, cost, currency, cost_breakdown, "
+                        "pricing_snapshot, quality, degraded, degraded_reason, created_at, updated_at) "
+                        "VALUES (:commodity, :start, :end, :quantity, :cost, 'EUR', '{}', '{}', 'valid', "
+                        "1, 'core test', :now, :now)"
+                    ),
+                    {**values, "commodity": commodity, "end": end},
+                )
+        with pytest.raises(IntegrityError):
+            connection.execute(
+                text(
+                    "INSERT INTO meter_cost_period "
+                    "(commodity, period_start, period_end, quantity, cost, currency, cost_breakdown, "
+                    "pricing_snapshot, quality, degraded, degraded_reason, created_at, updated_at) "
+                    "VALUES ('missing-reason', :start, '2026-08-23 00:15:00.000000', :quantity, :cost, "
+                    "'EUR', '{}', '{}', 'valid', 1, NULL, :now, :now)"
+                ),
+                values,
+            )
+    engine.dispose()
+
+
+def test_meter_cost_period_core_uses_utc_instant_idempotency_and_reloads_aware(tmp_path: Path) -> None:
+    database_url = f"sqlite:///{tmp_path / 'core-utc.db'}"
+    command.upgrade(_config(database_url), "head")
+    engine = create_engine(database_url)
+    utc_start = datetime(2026, 8, 22, 12, tzinfo=UTC)
+    values = {
+        "commodity": "heating",
+        "period_start": utc_start,
+        "period_end": utc_start + timedelta(minutes=15),
+        "quantity": Decimal("0.000000"),
+        "cost": Decimal("0.000000000"),
+        "currency": "EUR",
+        "cost_breakdown": {},
+        "pricing_snapshot": {},
+        "quality": "valid",
+        "degraded": True,
+        "degraded_reason": "core fixture",
+        "created_at": utc_start,
+        "updated_at": utc_start,
+    }
+    with engine.begin() as connection:
+        row_id = connection.execute(insert(MeterCostPeriod.__table__).values(values)).inserted_primary_key[0]
+    with engine.begin() as connection, pytest.raises(IntegrityError):
+        connection.execute(
+            insert(MeterCostPeriod.__table__).values(
+                {
+                    **values,
+                    "period_start": datetime(2026, 8, 22, 14, tzinfo=timezone(timedelta(hours=2))),
+                    "period_end": datetime(2026, 8, 22, 14, 15, tzinfo=timezone(timedelta(hours=2))),
+                }
+            )
+        )
+    with engine.connect() as connection, pytest.raises(StatementError, match="timezone-aware"):
+        connection.execute(
+            insert(MeterCostPeriod.__table__).values(
+                {**values, "commodity": "naive", "period_start": datetime(2026, 8, 23, 12)}
+            )
+        )
+    with Session(engine) as session:
+        stored = session.get(MeterCostPeriod, row_id)
+        assert stored is not None
+        assert stored.period_start == utc_start
+        assert stored.period_start.tzinfo is not None
+        assert stored.period_start.utcoffset() == timedelta(0)
+        stored.quality = "unverifiable"
+        session.commit()
+    engine.dispose()
+
+
+def test_meter_cost_period_foreign_keys_restrict_deletion(tmp_path: Path) -> None:
+    database_url = f"sqlite:///{tmp_path / 'foreign_keys.db'}"
+    command.upgrade(_config(database_url), "head")
+    engine = _fk_engine(database_url)
+    timestamp = datetime(2026, 8, 22, 12)
+    with engine.begin() as connection:
+        meter_id = connection.execute(
+            text(
+                "INSERT INTO meter (uuid, label, commodity, started_at, ended_at, reason, note, created_at) "
+                "VALUES ('meter-cost-test', 'Test', 'heating', :timestamp, NULL, 'initial', NULL, :timestamp)"
+            ),
+            {"timestamp": timestamp},
+        ).lastrowid
+        source_id = connection.execute(
+            text(
+                "INSERT INTO meter_source (uuid, name, kind, enabled, config, status, created_at, updated_at) "
+                "VALUES ('source-cost-test', 'Test', 'warmtelink_serial', 1, '{}', 'online', :timestamp, :timestamp)"
+            ),
+            {"timestamp": timestamp},
+        ).lastrowid
+        channel_id = connection.execute(
+            text(
+                "INSERT INTO meter_source_channel (uuid, source_id, channel_key, label, unit, created_at, updated_at) "
+                "VALUES ('channel-cost-test', :source_id, 'heating', 'Heating', 'GJ', :timestamp, :timestamp)"
+            ),
+            {"source_id": source_id, "timestamp": timestamp},
+        ).lastrowid
+        binding_id = connection.execute(
+            text(
+                "INSERT INTO meter_source_binding "
+                "(uuid, meter_id, channel_id, started_at, ended_at, created_at, updated_at) "
+                "VALUES ('binding-cost-test', :meter_id, :channel_id, :timestamp, NULL, :timestamp, :timestamp)"
+            ),
+            {"meter_id": meter_id, "channel_id": channel_id, "timestamp": timestamp},
+        ).lastrowid
+        contract_id = connection.execute(
+            text(
+                "INSERT INTO energy_contract (name, kind, scope, active, currency, created_at, updated_at) "
+                "VALUES ('Test', 'district_heating', 'thermal', 1, 'EUR', :timestamp, :timestamp)"
+            ),
+            {"timestamp": timestamp},
+        ).lastrowid
+        version_id = connection.execute(
+            text(
+                "INSERT INTO energy_contract_version "
+                "(contract_id, effective_from, effective_to, \"values\", created_at) "
+                "VALUES (:contract_id, :timestamp, NULL, '{}', :timestamp)"
+            ),
+            {"contract_id": contract_id, "timestamp": timestamp},
+        ).lastrowid
+        connection.execute(
+            text(
+                "INSERT INTO meter_cost_period "
+                "(commodity, period_start, period_end, meter_id, source_binding_id, contract_version_id, "
+                "quantity, cost, currency, cost_breakdown, pricing_snapshot, quality, degraded, degraded_reason, "
+                "created_at, updated_at) VALUES "
+                "('heating', :timestamp, :period_end, :meter_id, :binding_id, :version_id, "
+                "'0.001000', '0.010000000', 'EUR', '{}', '{}', 'valid', 0, NULL, :timestamp, :timestamp)"
+            ),
+            {
+                "timestamp": timestamp,
+                "period_end": timestamp + timedelta(minutes=15),
+                "meter_id": meter_id,
+                "binding_id": binding_id,
+                "version_id": version_id,
+            },
+        )
+        for statement, values in (
+            ("DELETE FROM meter WHERE id = :id", {"id": meter_id}),
+            ("DELETE FROM meter_source_binding WHERE id = :id", {"id": binding_id}),
+            ("DELETE FROM energy_contract_version WHERE id = :id", {"id": version_id}),
+        ):
+            with pytest.raises(IntegrityError):
+                connection.execute(text(statement), values)
+    engine.dispose()