diff --git a/src/akkudoktoreos/devices/settings/batterysettings.py b/src/akkudoktoreos/devices/settings/batterysettings.py
index 841f68df..0e84a67c 100644
--- a/src/akkudoktoreos/devices/settings/batterysettings.py
+++ b/src/akkudoktoreos/devices/settings/batterysettings.py
@@ -11,6 +11,7 @@ from numpydantic import NDArray, Shape
 from pydantic import Field, computed_field, field_validator, model_validator
 
 from akkudoktoreos.devices.settings.devicebasesettings import DevicesBaseSettings
+from akkudoktoreos.measurement.batterycapacity import BatteryCapacityEstimate, BatteryCapacityEstimationSettings
 
 if TYPE_CHECKING:
     from akkudoktoreos.devices.genetic0.genetic0battery import (
@@ -38,6 +39,12 @@ class BatteriesCommonSettings(DevicesBaseSettings):
     Note: Used for the GENETIC and GENETIC0 algorithm.
     """
 
+    capacity_estimation: Optional[BatteryCapacityEstimationSettings] = None
+    capacity_estimate: Optional[BatteryCapacityEstimate] = Field(
+        default=None,
+        description="Last explicitly stored capacity estimate; does not override capacity_wh.",
+    )
+
     capacity_wh: int = Field(
         default=8000,
         gt=0,
diff --git a/src/akkudoktoreos/measurement/batterycapacity.py b/src/akkudoktoreos/measurement/batterycapacity.py
new file mode 100644
index 00000000..454e0311
--- /dev/null
+++ b/src/akkudoktoreos/measurement/batterycapacity.py
@@ -0,0 +1,207 @@
+"""Estimate effective model capacity from independent SoC anchors and DC power."""
+
+from datetime import datetime
+from math import isfinite
+from typing import TYPE_CHECKING, Literal
+
+from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, model_validator
+
+from akkudoktoreos.config.configabc import SettingsBaseModel
+from akkudoktoreos.measurement.quality import SampleQuality
+
+if TYPE_CHECKING:
+    from akkudoktoreos.measurement.measurement import MeasurementChannelSettings
+
+
+class BatteryCapacityEstimationSettings(SettingsBaseModel):
+    """Only a battery-terminal DC power channel is supported, not inverter AC."""
+
+    power_key: str = Field(min_length=1)
+    positive_power: Literal["charging", "discharging"]
+    measurement_boundary: Literal["battery_dc"] = "battery_dc"
+    min_soc_change_percentage: float = Field(default=20, gt=0, le=100)
+    max_duration_hours: float = Field(default=168, gt=0, le=744)
+
+
+class CapacityModel(BaseModel):
+    model_config = ConfigDict(extra="forbid", allow_inf_nan=False)
+
+
+class BatteryCapacityRequest(CapacityModel):
+    """Caller attests both anchors are independent of the capacity being fitted.
+
+    End is the first confirmed full point when using the default end SoC.
+    Voltage/current anchors must be independently established for the chemistry.
+    """
+
+    start: AwareDatetime
+    end: AwareDatetime
+    start_soc_percentage: float = Field(ge=0, le=100)
+    end_soc_percentage: float = Field(default=100, ge=0, le=100)
+    soc_reference: Literal["bms", "external_calibration", "voltage_current_anchor"]
+    store_estimate: bool = False
+
+    @model_validator(mode="after")
+    def ordered(self):
+        if self.end.timestamp() <= self.start.timestamp():
+            raise ValueError("end must be after start.")
+        return self
+
+
+class BatteryCapacityEstimate(CapacityModel):
+    battery_id: str
+    start: AwareDatetime
+    end: AwareDatetime
+    soc_reference: str
+    start_soc_percentage: float
+    end_soc_percentage: float
+    estimated_capacity_wh: float
+    configured_capacity_wh: float
+    capacity_change_percentage: float
+    model_end_soc_percentage_unclipped: float
+    model_soc_error_percentage_points: float
+    charge_energy_wh: float
+    discharge_energy_wh: float
+    stored_energy_change_wh: float
+    charging_efficiency: float
+    discharging_efficiency: float
+    power_key: str
+    positive_power: str
+    integration_method: str
+    coverage_seconds: float
+    samples_used: int
+    warnings: list[str]
+
+
+def estimate_capacity(
+    request: BatteryCapacityRequest,
+    settings: BatteryCapacityEstimationSettings,
+    channel: "MeasurementChannelSettings",
+    samples: list[tuple[datetime, float | None, SampleQuality]],
+    *,
+    battery_id: str,
+    capacity_wh: float,
+    charging_efficiency: float,
+    discharging_efficiency: float,
+) -> BatteryCapacityEstimate:
+    """Fit C in C * delta_soc = eta_c * E_charge - E_discharge / eta_d.
+
+    No clipping at 0/100%, extrapolation, gap filling or efficiency fitting.
+    Linear segments are split at zero before applying directional efficiencies.
+    """
+    left, right = request.start.timestamp(), request.end.timestamp()
+    if right - left > settings.max_duration_hours * 3600:
+        raise ValueError("Requested period exceeds capacity_estimation.max_duration_hours.")
+    delta_soc = (request.end_soc_percentage - request.start_soc_percentage) / 100
+    if abs(delta_soc) * 100 + 1e-9 < settings.min_soc_change_percentage:
+        raise ValueError(
+            "Independent SoC change is too small; full-to-full cannot identify capacity."
+        )
+    if channel.quantity != "power":
+        raise ValueError("Capacity estimation requires signed battery DC power in W or kW.")
+    if not isfinite(capacity_wh) or capacity_wh <= 0:
+        raise ValueError("Configured capacity must be positive and finite.")
+    if not all(isfinite(e) and 0 < e <= 1 for e in (charging_efficiency, discharging_efficiency)):
+        raise ValueError("Battery efficiencies must be finite and in (0, 1].")
+    points = []
+    for time, value, quality in samples:
+        if time.tzinfo is None or time.utcoffset() is None:
+            raise ValueError("Explicit sample timezone required.")
+        points.append((time.timestamp(), value, quality))
+    points.sort(key=lambda row: row[0])
+    if any(a[0] == b[0] for a, b in zip(points, points[1:])):
+        raise ValueError("Duplicate power sample timestamps.")
+    factor = (1000 if channel.unit == "kW" else 1) * (
+        1 if settings.positive_power == "charging" else -1
+    )
+    charge = discharge = coverage = 0.0
+    cumulative_net = 0.0
+    minimum_net = maximum_net = 0.0
+    used = set()
+    cursor = left
+    for (a, va, qa), (b, vb, qb) in zip(points, points[1:]):
+        lo, hi = max(left, a), min(right, b)
+        if hi <= lo:
+            continue
+        if lo > cursor + 1e-6 or b - a > channel.max_gap_seconds:
+            raise ValueError("Incomplete power coverage or sample gap exceeds max_gap_seconds.")
+        if (
+            any(v is None or isinstance(v, bool) or not isfinite(v) for v in (va, vb))
+            or qa.status != "measured"
+            or qb.status != "measured"
+        ):
+            raise ValueError(
+                "Capacity estimation requires finite measured power samples throughout."
+            )
+        if qb.reset or qa.generation != qb.generation:
+            raise ValueError("Power sensor reset or generation change within the period.")
+        p, q = va * factor, vb * factor
+        if channel.integration_method == "linear":
+            slope = (q - p) / (b - a)
+            p, q = p + slope * (lo - a), p + slope * (hi - a)
+        else:
+            q = p
+        duration = hi - lo
+        if p * q < 0:
+            first = duration * abs(p) / (abs(p) + abs(q))
+            parts = [(p * first / 7200), (q * (duration - first) / 7200)]
+        else:
+            parts = [(p + q) * duration / 7200]
+        charge += sum(max(0, energy) for energy in parts)
+        discharge += sum(max(0, -energy) for energy in parts)
+        for energy in parts:
+            cumulative_net += (
+                energy * charging_efficiency if energy >= 0 else energy / discharging_efficiency
+            )
+            minimum_net = min(minimum_net, cumulative_net)
+            maximum_net = max(maximum_net, cumulative_net)
+        coverage += duration
+        cursor = hi
+        used.update((a, b))
+    if cursor < right - 1e-6 or abs(coverage - (right - left)) > 1e-6:
+        raise ValueError("Incomplete power coverage; no extrapolation to the SoC anchors.")
+    net = charge * charging_efficiency - discharge / discharging_efficiency
+    capacity = net / delta_soc
+    if not isfinite(capacity) or capacity <= 0:
+        raise ValueError("Energy flow disagrees with SoC change; check polarity and anchors.")
+    if (
+        request.start_soc_percentage + minimum_net / capacity * 100 < -1e-6
+        or request.start_soc_percentage + maximum_net / capacity * 100 > 100 + 1e-6
+    ):
+        raise ValueError(
+            "Fitted SoC leaves 0..100% within the period; check anchors and use the first full point."
+        )
+    model_end = request.start_soc_percentage + net / capacity_wh * 100
+    warnings = [
+        "Conditional estimate: SoC anchors, DC measurement boundary and configured efficiencies must be correct.",
+        "A single interval cannot independently identify both capacity and efficiencies.",
+        "The active capacity_wh is unchanged.",
+    ]
+    if request.soc_reference == "voltage_current_anchor":
+        warnings.append(
+            "Voltage/current anchors depend on chemistry, temperature and operating conditions."
+        )
+    return BatteryCapacityEstimate(
+        battery_id=battery_id,
+        start=request.start,
+        end=request.end,
+        soc_reference=request.soc_reference,
+        start_soc_percentage=request.start_soc_percentage,
+        end_soc_percentage=request.end_soc_percentage,
+        estimated_capacity_wh=capacity,
+        configured_capacity_wh=capacity_wh,
+        capacity_change_percentage=(capacity / capacity_wh - 1) * 100,
+        model_end_soc_percentage_unclipped=model_end,
+        model_soc_error_percentage_points=model_end - request.end_soc_percentage,
+        charge_energy_wh=charge,
+        discharge_energy_wh=discharge,
+        stored_energy_change_wh=net,
+        charging_efficiency=charging_efficiency,
+        discharging_efficiency=discharging_efficiency,
+        power_key=settings.power_key,
+        positive_power=settings.positive_power,
+        integration_method=channel.integration_method,
+        coverage_seconds=coverage,
+        samples_used=len(used),
+        warnings=warnings,
+    )
diff --git a/src/akkudoktoreos/measurement/energy.py b/src/akkudoktoreos/measurement/energy.py
new file mode 100644
index 00000000..df610a19
--- /dev/null
+++ b/src/akkudoktoreos/measurement/energy.py
@@ -0,0 +1,190 @@
+"""Quality-aware energy conversion of raw measurement samples.
+
+No extrapolation beyond the last sample. Intervals use elapsed UTC seconds.
+"""
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from math import isfinite
+from typing import TYPE_CHECKING, Iterable, Literal
+
+from akkudoktoreos.measurement.quality import SampleQuality
+
+if TYPE_CHECKING:
+    from akkudoktoreos.measurement.measurement import MeasurementChannelSettings
+
+
+@dataclass(frozen=True)
+class EnergyInterval:
+    """Energy and actual temporal support; missing energy is never zero-filled."""
+
+    start: datetime
+    end: datetime
+    energy_wh: float | None
+    observed_energy_wh: float | None
+    coverage_seconds: float
+    coverage_status: Literal["complete", "partial", "missing", "invalid"]
+    methods: tuple[str, ...]
+    flags: tuple[str, ...]
+    # Retain the support for subsequent multi-channel balance intersection.
+    coverage_ranges: tuple[tuple[datetime, datetime], ...]
+
+
+def _timestamp(value: datetime) -> float:
+    if value.tzinfo is None or value.utcoffset() is None:
+        raise ValueError("Explicit timezone required.")
+    return value.timestamp()
+
+
+def _date(value: float) -> datetime:
+    return datetime.fromtimestamp(value, timezone.utc)
+
+
+def energy_intervals(
+    samples: Iterable[tuple[datetime, float | None]],
+    channel: "MeasurementChannelSettings",
+    start: datetime,
+    end: datetime,
+    interval_seconds: int = 900,
+    quality: dict[float, SampleQuality] | None = None,
+) -> list[EnergyInterval]:
+    """Convert finite samples, preserving gaps, resets and partial coverage.
+
+    Hold applies only between consecutive observations within max_gap_seconds.
+    Null/invalid observations break continuity. Meter resets invalidate that segment.
+    Fixed interval energy is uniformly allocated when a target cuts its source interval.
+    """
+    left, right = _timestamp(start), _timestamp(end)
+    if right <= left or type(interval_seconds) is not int or interval_seconds <= 0:
+        raise ValueError("Require end > start and positive integer interval_seconds.")
+    points = sorted(((_timestamp(t), v) for t, v in samples), key=lambda point: point[0])
+    if any(a[0] == b[0] for a, b in zip(points, points[1:])):
+        raise ValueError("Duplicate sample timestamps must be resolved before conversion.")
+    scale = 1000 if channel.unit in ("kW", "kWh") else 1
+    quality = quality or {}
+
+    def sample_quality(time: float) -> SampleQuality:
+        return quality.get(time, SampleQuality())
+
+    def number(value: float | None) -> float | None:
+        if isinstance(value, bool) or not isinstance(value, (int, float)) or not isfinite(value):
+            return None
+        return float(value) * scale
+
+    # Segment: start, end, start/end power in W, method, error flag.
+    segments = []
+    if channel.quantity == "interval_energy":
+        duration = channel.interval_seconds
+        if duration is None:
+            raise ValueError("Interval duration is required.")
+        previous_end = None
+        for time, raw in points:
+            a = time if channel.timestamp_reference == "start" else time - duration
+            b = a + duration
+            if previous_end is not None and a < previous_end:
+                raise ValueError("Overlapping source energy intervals.")
+            previous_end = b
+            value = number(raw)
+            if sample_quality(time).status in ("invalid", "unavailable"):
+                value = None
+            power = value * 3600 / duration if value is not None else None
+            segments.append(
+                (a, b, power, power, "interval_energy", "invalid_sample" if power is None else None)
+            )
+    else:
+        for (a, raw_a), (b, raw_b) in zip(points, points[1:]):
+            va, vb = number(raw_a), number(raw_b)
+            qa, qb = sample_quality(a), sample_quality(b)
+            if qa.status in ("invalid", "unavailable"):
+                va = None
+            if qb.status in ("invalid", "unavailable"):
+                vb = None
+            flag = None
+            if channel.max_gap_seconds is not None and b - a > channel.max_gap_seconds:
+                flag = "gap_too_large"
+            elif va is None or (
+                vb is None
+                and (
+                    channel.quantity == "cumulative_energy"
+                    or channel.integration_method == "linear"
+                )
+            ):
+                flag = "invalid_sample"
+            if channel.quantity == "cumulative_energy":
+                method = "meter_difference"
+                if va is not None and vb is not None and vb < va:
+                    flag = "meter_reset"
+                if qb.reset or qa.generation != qb.generation:
+                    flag = "meter_reset"
+                pa = pb = (
+                    (vb - va) * 3600 / (b - a)
+                    if flag is None and va is not None and vb is not None
+                    else None
+                )
+            else:
+                method = "integrated_power"
+                pa = va
+                pb = vb if channel.integration_method == "linear" else va
+            segments.append((a, b, pa, pb, method, flag))
+
+    result = []
+    slot = left
+    segment_index = 0
+    while slot < right:
+        stop = min(slot + interval_seconds, right)
+        total, coverage = 0.0, 0.0
+        methods: set[str] = set()
+        flags: set[str] = set()
+        ranges: list[tuple[datetime, datetime]] = []
+        while segment_index < len(segments) and segments[segment_index][1] <= slot:
+            segment_index += 1
+        for index in range(segment_index, len(segments)):
+            a, b, pa, pb, method, flag = segments[index]
+            if a >= stop:
+                break
+            lo, hi = max(a, slot), min(b, stop)
+            if hi <= lo:
+                continue
+            # Quality follows the endpoints used by the integration rule.
+            source_time = a
+            if channel.quantity == "interval_energy" and channel.timestamp_reference == "end":
+                source_time = b
+            statuses = {sample_quality(source_time).status}
+            if channel.quantity == "cumulative_energy" or channel.integration_method == "linear":
+                statuses.add(sample_quality(b).status)
+            flags.update(s for s in statuses if s != "measured")
+            if flag is not None:
+                flags.add(flag)
+                continue
+            if pa is None or pb is None:
+                raise ValueError("Valid segment requires finite endpoint powers.")
+            p_lo = pa + (pb - pa) * (lo - a) / (b - a)
+            p_hi = pa + (pb - pa) * (hi - a) / (b - a)
+            total += (p_lo + p_hi) / 2 * (hi - lo) / 3600
+            coverage += hi - lo
+            if ranges and ranges[-1][1] == _date(lo):
+                ranges[-1] = (ranges[-1][0], _date(hi))
+            else:
+                ranges.append((_date(lo), _date(hi)))
+            methods.add(method)
+            if method in ("meter_difference", "interval_energy") and (lo != a or hi != b):
+                methods.add("allocated_energy")
+        complete = abs(coverage - (stop - slot)) < 1e-6
+        status: Literal["complete", "partial", "missing", "invalid"] = (
+            "complete" if complete else "partial" if coverage else "invalid" if flags else "missing"
+        )
+        result.append(
+            EnergyInterval(
+                _date(slot),
+                _date(stop),
+                total if complete else None,
+                total if coverage else None,
+                coverage,
+                status,
+                tuple(sorted(methods)),
+                tuple(sorted(flags)),
+                tuple(ranges),
+            )
+        )
+        slot = stop
+    return result
diff --git a/src/akkudoktoreos/measurement/household.py b/src/akkudoktoreos/measurement/household.py
new file mode 100644
index 00000000..27358d6c
--- /dev/null
+++ b/src/akkudoktoreos/measurement/household.py
@@ -0,0 +1,118 @@
+"""AC household balances on the intersection of actual measurement support."""
+
+from datetime import datetime
+from typing import Callable, Literal
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from akkudoktoreos.measurement.energy import EnergyInterval
+
+
+class HouseholdInput(BaseModel):
+    """A non-overlapping AC branch; polarity normalizes the sensor's sign."""
+
+    model_config = ConfigDict(extra="forbid")
+    key: str
+    branch: str = Field(min_length=1)
+    role: Literal["site", "grid", "pv", "battery", "inverter", "ev", "device"]
+    polarity: Literal[-1, 1] = 1
+
+
+class HouseholdSettings(BaseModel):
+    """Fixed topology, never user-supplied executable balance expressions.
+
+    Grid import, PV production and battery/inverter discharge are positive.
+    EV/device inputs are positive consumption, subtracted only from derived loads.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+    topology: Literal["direct", "separate_ac", "hybrid_ac"]
+    inputs: list[HouseholdInput] = Field(min_length=1)
+
+    @model_validator(mode="after")
+    def validate_topology(self) -> "HouseholdSettings":
+        keys = [item.key for item in self.inputs]
+        branches = [item.branch for item in self.inputs]
+        if len(keys) != len(set(keys)) or len(branches) != len(set(branches)):
+            raise ValueError("Duplicate measurement key or physical branch in household balance.")
+        roles = {item.role for item in self.inputs} - {"ev", "device"}
+        allowed = {
+            "direct": {"site"},
+            "separate_ac": {"grid", "pv", "battery"},
+            # Additional AC PV branches are independent of the hybrid's net output.
+            # Its DC PV/battery must not be added again as separate inputs.
+            "hybrid_ac": {"grid", "inverter", "pv"},
+        }[self.topology]
+        required = {"site"} if self.topology == "direct" else {"grid"}
+        if self.topology == "hybrid_ac":
+            required.add("inverter")
+        if not required <= roles or not roles <= allowed:
+            raise ValueError("Inputs do not match the selected AC topology.")
+        return self
+
+
+def household_intervals(
+    settings: HouseholdSettings,
+    convert: Callable[[str, datetime, datetime, int], list[EnergyInterval]],
+    start: datetime,
+    end: datetime,
+    interval_seconds: int = 900,
+) -> dict[str, list[EnergyInterval]]:
+    """Integrate each source again over shared support, never prorate partial sums."""
+    series = {item.key: convert(item.key, start, end, interval_seconds) for item in settings.inputs}
+    output: dict[str, list[EnergyInterval]] = {}
+    for name, excluded in (
+        ("site", {"ev", "device"}),
+        ("household", {"device"}),
+        ("base", set()),
+    ):
+        inputs = [item for item in settings.inputs if item.role not in excluded]
+        result = []
+        for index, template in enumerate(series[inputs[0].key]):
+            rows = [series[item.key][index] for item in inputs]
+            # Partition at every coverage boundary; keep only the intersection.
+            boundaries = sorted({t for row in rows for pair in row.coverage_ranges for t in pair})
+            support = [
+                (a, b)
+                for a, b in zip(boundaries, boundaries[1:])
+                if all(any(lo <= a and b <= hi for lo, hi in row.coverage_ranges) for row in rows)
+            ]
+            total = 0.0
+            methods = {method for row in rows for method in row.methods}
+            flags = {flag for row in rows for flag in row.flags}
+            for a, b in support:
+                for item in inputs:
+                    # The range is at most one target interval, but may be fractional seconds.
+                    parts = convert(item.key, a, b, interval_seconds)
+                    value = parts[0].energy_wh
+                    if value is None:
+                        raise ValueError("Inconsistent measurement support during balance.")
+                    sign = -1 if item.role in ("ev", "device") else 1
+                    total += sign * item.polarity * value
+                    methods.update(parts[0].methods)
+                    flags.update(parts[0].flags)
+            coverage = sum((b - a).total_seconds() for a, b in support)
+            complete = abs(coverage - (template.end - template.start).total_seconds()) < 1e-6
+            if coverage and total < -1e-6:
+                flags.add("negative_balance")
+            result.append(
+                EnergyInterval(
+                    template.start,
+                    template.end,
+                    total if complete else None,
+                    total if coverage else None,
+                    coverage,
+                    "complete"
+                    if complete
+                    else "partial"
+                    if coverage
+                    else "invalid"
+                    if flags
+                    else "missing",
+                    tuple(sorted(methods | {"ac_balance"})),
+                    tuple(sorted(flags)),
+                    tuple(support),
+                )
+            )
+        output[name] = result
+    return output
diff --git a/src/akkudoktoreos/measurement/measurement.py b/src/akkudoktoreos/measurement/measurement.py
index 6cd25415..63098c04 100644
--- a/src/akkudoktoreos/measurement/measurement.py
+++ b/src/akkudoktoreos/measurement/measurement.py
@@ -6,17 +6,29 @@ data records for measurements.
 The measurements can be added programmatically or imported from a file or JSON string.
 """
 
+import json
+from bisect import bisect_left, bisect_right
+from datetime import datetime, timedelta
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any, Callable, Literal, Optional
 
 import numpy as np
 from loguru import logger
 from numpydantic import NDArray, Shape
-from pydantic import Field, computed_field
+from pydantic import Field, computed_field, model_validator
 
 from akkudoktoreos.config.configabc import SettingsBaseModel
 from akkudoktoreos.core.coreabc import SingletonMixin
 from akkudoktoreos.core.dataabc import DataImportMixin, DataRecord, DataSequence
+from akkudoktoreos.core.databaseabc import DatabaseTimestamp
+from akkudoktoreos.measurement.energy import EnergyInterval
+from akkudoktoreos.measurement.batterycapacity import (
+    BatteryCapacityEstimate,
+    BatteryCapacityRequest,
+    estimate_capacity,
+)
+from akkudoktoreos.measurement.household import HouseholdSettings, household_intervals
+from akkudoktoreos.measurement.quality import MeasurementSample, SampleQuality
 from akkudoktoreos.utils.datetimeutil import (
     DateTime,
     Duration,
@@ -25,6 +37,41 @@ from akkudoktoreos.utils.datetimeutil import (
 )
 
 
+class MeasurementChannelSettings(SettingsBaseModel):
+    """Meaning of a raw measurement channel; no conversion is performed on storage."""
+
+    quantity: Literal["power", "cumulative_energy", "interval_energy"]
+    unit: Literal["W", "kW", "Wh", "kWh"]
+    integration_method: Optional[Literal["hold", "linear"]] = None
+    max_gap_seconds: Optional[int] = Field(default=None, gt=0, strict=True)
+    interval_seconds: Optional[int] = Field(default=None, gt=0, strict=True)
+    timestamp_reference: Optional[Literal["start", "end"]] = None
+
+    @model_validator(mode="after")
+    def validate_semantics(self) -> "MeasurementChannelSettings":
+        """Reject ambiguous units and time semantics before accepting a channel."""
+        if self.quantity == "power":
+            if self.unit not in ("W", "kW"):
+                raise ValueError("Power channels require W or kW.")
+            if self.integration_method is None or self.max_gap_seconds is None:
+                raise ValueError("Power channels require integration_method and max_gap_seconds.")
+        else:
+            if self.unit not in ("Wh", "kWh"):
+                raise ValueError("Energy channels require Wh or kWh.")
+            if self.integration_method is not None:
+                raise ValueError("integration_method is only applicable to power channels.")
+        if self.quantity == "interval_energy":
+            if self.interval_seconds is None or self.timestamp_reference is None:
+                raise ValueError(
+                    "Interval energy requires interval_seconds and timestamp_reference."
+                )
+            if self.max_gap_seconds is not None:
+                raise ValueError("Interval energy uses explicit intervals, not max_gap_seconds.")
+        elif self.interval_seconds is not None or self.timestamp_reference is not None:
+            raise ValueError("Interval metadata is only applicable to interval_energy channels.")
+        return self
+
+
 class MeasurementCommonSettings(SettingsBaseModel):
     """Measurement Configuration."""
 
@@ -37,6 +84,41 @@ class MeasurementCommonSettings(SettingsBaseModel):
         },
     )
 
+    channels: dict[str, MeasurementChannelSettings] = Field(
+        default_factory=dict,
+        json_schema_extra={
+            "description": "Typed raw measurement channels keyed by measurement key."
+        },
+    )
+
+    household: Optional[HouseholdSettings] = None
+    energy_context_seconds: int = Field(default=86400, gt=0, le=604800, strict=True)
+
+    @model_validator(mode="after")
+    def validate_channels(self) -> "MeasurementCommonSettings":
+        """Preserve the kWh meter contract of legacy keys and avoid record collisions."""
+        for key, channel in self.channels.items():
+            if (
+                not key
+                or key != key.strip()
+                or key.startswith("_")
+                or hasattr(DataRecord, key)
+                or key in DataRecord.model_fields
+                or key == "sample_quality"
+            ):
+                raise ValueError(f"Invalid or reserved measurement channel key: {key!r}")
+            for name in type(self).model_fields:
+                if name.endswith("_emr_keys") and key in (getattr(self, name) or []):
+                    if channel.quantity != "cumulative_energy" or channel.unit != "kWh":
+                        raise ValueError(
+                            f"Legacy meter key {key!r} must remain cumulative_energy in kWh."
+                        )
+        if self.household is not None:
+            for item in self.household.inputs:
+                if item.key not in self.keys:
+                    raise ValueError(f"Unknown household measurement key: {item.key!r}")
+        return self
+
     load_emr_keys: Optional[list[str]] = Field(
         default=None,
         json_schema_extra={
@@ -74,7 +156,7 @@ class MeasurementCommonSettings(SettingsBaseModel):
     @property
     def keys(self) -> list[str]:
         """The keys of the measurements that can be stored."""
-        key_list = []
+        key_list = list(self.channels)
         for key in self.__class__.model_fields.keys():
             if key.endswith("_keys") and (value := getattr(self, key)):
                 key_list.extend(value)
@@ -84,6 +166,18 @@ class MeasurementCommonSettings(SettingsBaseModel):
 class MeasurementDataRecord(DataRecord):
     """Represents a measurement data record containing various measurements at a specific datetime."""
 
+    sample_quality: dict[str, SampleQuality] = Field(default_factory=dict)
+
+    @classmethod
+    def record_keys(cls) -> list[str]:
+        """Quality is stored alongside values, not exposed as a numeric channel."""
+        return [key for key in super().record_keys() if key != "sample_quality"]
+
+    @classmethod
+    def record_keys_writable(cls) -> list[str]:
+        """Only the typed sample path writes quality, never numeric import paths."""
+        return [key for key in super().record_keys_writable() if key != "sample_quality"]
+
     @classmethod
     def configured_data_keys(cls) -> Optional[list[str]]:
         """Return the keys for the configured field like data."""
@@ -149,6 +243,176 @@ class Measurement(SingletonMixin, DataImportMixin, DataSequence[MeasurementDataR
         # Return ceiling of division to include partial intervals
         return int(np.ceil(diff_seconds / interval_seconds))
 
+    async def import_samples(self, samples: list[MeasurementSample]) -> None:
+        """Validate the whole batch before replacing samples; omitted keys stay untouched."""
+        for sample in samples:
+            self._energy_channel(sample.key)
+        for sample in samples:
+            dt = to_datetime(sample.date_time)
+            await self.update_value(dt, sample.key, sample.value)
+            record = await self.db_get_record(DatabaseTimestamp.from_datetime(dt))
+            if not isinstance(record, MeasurementDataRecord):
+                raise RuntimeError("Measurement sample was not stored.")
+            record.sample_quality[sample.key] = sample.quality.model_copy(deep=True)
+            await self.db_mark_dirty_record(record)
+
+    async def insert_by_datetime(self, record: DataRecord) -> None:
+        """Merge quality by channel as well as the ordinary measurement fields."""
+        await super().insert_by_datetime(record)
+        if (
+            isinstance(record, MeasurementDataRecord)
+            and record.sample_quality
+            and record.date_time is not None
+        ):
+            stored = await self.db_get_record(DatabaseTimestamp.from_datetime(record.date_time))
+            if not isinstance(stored, MeasurementDataRecord):
+                raise RuntimeError("Measurement record was not stored.")
+            stored.sample_quality = stored.sample_quality | record.sample_quality
+            await self.db_mark_dirty_record(stored)
+
+    def _energy_channel(self, key: str) -> MeasurementChannelSettings:
+        channel = self.config.measurement.channels.get(key)
+        if channel is not None:
+            return channel
+        for name in type(self.config.measurement).model_fields:
+            if name.endswith("_emr_keys") and key in (getattr(self.config.measurement, name) or []):
+                return MeasurementChannelSettings(quantity="cumulative_energy", unit="kWh")
+        raise ValueError(f"No energy channel definition for {key!r}.")
+
+    async def _energy_converter(
+        self,
+        keys: list[str],
+        start: datetime,
+        end: datetime,
+    ) -> Callable[[str, datetime, datetime, int], list[EnergyInterval]]:
+        """Load a bounded window once, including configured boundary context."""
+        from akkudoktoreos.measurement.energy import energy_intervals
+
+        if any(dt.tzinfo is None or dt.utcoffset() is None for dt in (start, end)):
+            raise ValueError("Explicit timezone required.")
+        seconds = end.timestamp() - start.timestamp()
+        if not 0 < seconds <= 31 * 86400:
+            raise ValueError("Energy queries require a positive range of at most 31 days.")
+        channels = {key: self._energy_channel(key) for key in keys}
+        context = self.config.measurement.energy_context_seconds
+        records = [record async for record in self.db_iterate_records(
+                DatabaseTimestamp.from_datetime(to_datetime(start) - timedelta(seconds=context)),
+                DatabaseTimestamp.from_datetime(
+                    to_datetime(end) + timedelta(seconds=context, microseconds=1)
+                ),
+            )
+        ]
+        samples = {
+            key: [
+                (record.date_time, record.configured_data[key])
+                for record in records
+                if key in record.configured_data and record.date_time is not None
+            ]
+            for key in keys
+        }
+        quality = {
+            key: {
+                record.date_time.timestamp(): record.sample_quality[key]
+                for record in records
+                if key in record.sample_quality and record.date_time is not None
+            }
+            for key in keys
+        }
+        timestamps = {key: [dt.timestamp() for dt, _ in values] for key, values in samples.items()}
+        for key, channel in channels.items():
+            duration = channel.interval_seconds
+            if (
+                channel.quantity == "interval_energy"
+                and duration is not None
+                and any(b - a < duration for a, b in zip(timestamps[key], timestamps[key][1:]))
+            ):
+                raise ValueError(f"Overlapping source energy intervals for {key!r}.")
+
+        def convert(
+            key: str, left: datetime, right: datetime, interval_seconds: int
+        ) -> list[EnergyInterval]:
+            if type(interval_seconds) is not int or interval_seconds <= 0:
+                raise ValueError("Positive integer interval_seconds required.")
+            if (right.timestamp() - left.timestamp()) / interval_seconds > 10000:
+                raise ValueError("At most 10000 output intervals per query.")
+            lo = max(0, bisect_left(timestamps[key], left.timestamp()) - 1)
+            hi = bisect_right(timestamps[key], right.timestamp()) + 1
+            return energy_intervals(
+                samples[key][lo:hi], channels[key], left, right, interval_seconds, quality[key]
+            )
+
+        return convert
+
+    async def household_intervals(
+        self, start_datetime: datetime, end_datetime: datetime, interval_seconds: int = 900
+    ) -> dict[str, list[EnergyInterval]]:
+        """Return site, household without EV, and base without configured devices."""
+        settings = self.config.measurement.household
+        if settings is None:
+            raise ValueError("No household balance configured.")
+        # Revalidate references even after an in-place configuration mutation.
+        settings = HouseholdSettings.model_validate(settings.model_dump())
+        convert = await self._energy_converter(
+            [item.key for item in settings.inputs], start_datetime, end_datetime
+        )
+        return household_intervals(
+            settings, convert, start_datetime, end_datetime, interval_seconds
+        )
+
+    async def energy_intervals(
+        self,
+        key: str,
+        start_datetime: datetime,
+        end_datetime: datetime,
+        interval_seconds: int = 900,
+    ) -> list[EnergyInterval]:
+        """Convert a typed channel without changing the legacy kWh calculation.
+
+        Read only explicitly stored values for this key: another channel's timestamp
+        must not introduce a synthetic outage. Explicit null values remain barriers.
+        """
+        convert = await self._energy_converter([key], start_datetime, end_datetime)
+        return convert(key, start_datetime, end_datetime, interval_seconds)
+
+    async def estimate_battery_capacity(
+        self, battery_id: str, request: BatteryCapacityRequest
+    ) -> BatteryCapacityEstimate:
+        """Read signed DC samples without mutating raw data or the active capacity."""
+        batteries = [b for b in (self.config.devices.batteries or {}).values() if b.device_id == battery_id]
+        if len(batteries) != 1:
+            raise ValueError("Require exactly one configured battery with this device_id.")
+        battery = batteries[0]
+        settings = battery.capacity_estimation
+        if settings is None:
+            raise ValueError("Configure devices.batteries[device_id].capacity_estimation first.")
+        if request.end.timestamp() - request.start.timestamp() > settings.max_duration_hours * 3600:
+            raise ValueError("Requested period exceeds capacity_estimation.max_duration_hours.")
+        channel = self._energy_channel(settings.power_key)
+        if channel.quantity != "power":
+            raise ValueError("Capacity estimation requires a battery DC power channel.")
+        context = channel.max_gap_seconds
+        samples = []
+        async for record in self.db_iterate_records(
+            DatabaseTimestamp.from_datetime(to_datetime(request.start) - timedelta(seconds=context)),
+            DatabaseTimestamp.from_datetime(
+                to_datetime(request.end) + timedelta(seconds=context, microseconds=1)
+            ),
+        ):
+            if record.date_time is not None and settings.power_key in record.configured_data:
+                samples.append((
+                    record.date_time,
+                    record.configured_data[settings.power_key],
+                    record.sample_quality.get(settings.power_key, SampleQuality()),
+                ))
+                if len(samples) > 250000:
+                    raise ValueError("More than 250000 power samples; request a shorter period.")
+        return estimate_capacity(
+            request, settings, channel, samples, battery_id=battery.device_id,
+            capacity_wh=battery.capacity_wh,
+            charging_efficiency=battery.charging_efficiency,
+            discharging_efficiency=battery.discharging_efficiency,
+        )
+
     async def _energy_from_meter_readings(
         self,
         key: str,
@@ -313,13 +577,12 @@ class Measurement(SingletonMixin, DataImportMixin, DataSequence[MeasurementDataR
             if not measurement_file_path.exists():
                 return False
             try:
-                # Validate into a temporary instance
-                loaded = self.__class__.model_validate_json(
-                    measurement_file_path.read_text(encoding="utf-8")
-                )
-
-                # Explicitly add data records to the existing singleton
-                for record in loaded.records:
+                # Measurement is a singleton: validating a temporary Measurement
+                # returns the existing instance and discards serialized records.
+                payload = json.loads(measurement_file_path.read_text(encoding="utf-8"))
+                records = [MeasurementDataRecord.model_validate(data)
+                           for data in payload.get("records", [])]
+                for record in records:
                     await self.insert_by_datetime(record)
             except Exception as e:
                 logger.exception("Cannot load measurements")
diff --git a/src/akkudoktoreos/measurement/quality.py b/src/akkudoktoreos/measurement/quality.py
new file mode 100644
index 00000000..cfb5dd06
--- /dev/null
+++ b/src/akkudoktoreos/measurement/quality.py
@@ -0,0 +1,24 @@
+"""Optional quality information on raw measurement samples."""
+
+from typing import Literal
+
+from pydantic import AwareDatetime, BaseModel, ConfigDict, Field
+
+
+class SampleQuality(BaseModel):
+    """Reset marks the first reading after a reset; generation identifies a meter."""
+
+    model_config = ConfigDict(extra="forbid")
+    status: Literal["measured", "estimated", "invalid", "unavailable"] = "measured"
+    reset: bool = False
+    generation: str | None = Field(default=None, max_length=128)
+
+
+class MeasurementSample(BaseModel):
+    """Complete replacement of one key/timestamp, including its quality."""
+
+    model_config = ConfigDict(extra="forbid", allow_inf_nan=False)
+    date_time: AwareDatetime
+    key: str
+    value: float | None = Field(strict=True)
+    quality: SampleQuality = Field(default_factory=SampleQuality)
diff --git a/src/akkudoktoreos/server/eos.py b/src/akkudoktoreos/server/eos.py
index 258ff2ba..1a08c878 100755
--- a/src/akkudoktoreos/server/eos.py
+++ b/src/akkudoktoreos/server/eos.py
@@ -75,6 +75,7 @@ from akkudoktoreos.prediction.loadakkudoktor import LoadAkkudoktorCommonSettings
 from akkudoktoreos.prediction.pvforecast import PVForecastCommonSettings
 from akkudoktoreos.prediction.pvforecastpvlib import _cec_inverters, _cec_modules
 from akkudoktoreos.server.container_healthcheck import publish_port
+from akkudoktoreos.server.rest.measurement import router as measurement_router
 from akkudoktoreos.server.rest.error import (
     EOSProblem,
     create_error_page,
@@ -261,6 +262,8 @@ The genetic optimization API fields were renamed from German to English. For bac
 # Application generic exception handling
 # ----------------------
 
+app.include_router(measurement_router)
+
 register_problem_handlers(app)
 
 
diff --git a/src/akkudoktoreos/server/rest/measurement.py b/src/akkudoktoreos/server/rest/measurement.py
new file mode 100644
index 00000000..145b5400
--- /dev/null
+++ b/src/akkudoktoreos/server/rest/measurement.py
@@ -0,0 +1,103 @@
+"""Typed samples and derived energy within the existing measurement API."""
+
+from typing import Annotated
+
+from fastapi import APIRouter, Body, HTTPException, Query
+from pydantic import AwareDatetime
+
+from akkudoktoreos.core.coreabc import get_config, get_measurement
+from akkudoktoreos.measurement.batterycapacity import BatteryCapacityEstimate, BatteryCapacityRequest
+from akkudoktoreos.core.databaseabc import DatabaseTimestamp
+from akkudoktoreos.measurement.energy import EnergyInterval
+from akkudoktoreos.measurement.quality import MeasurementSample, SampleQuality
+from akkudoktoreos.utils.datetimeutil import to_datetime
+
+router = APIRouter(prefix="/v1/measurement", tags=["measurement"])
+
+
+@router.post("/battery-capacity/{battery_id}", response_model=BatteryCapacityEstimate)
+async def post_battery_capacity(battery_id: str, request: BatteryCapacityRequest) -> BatteryCapacityEstimate:
+    """Estimate capacity from independent SoC anchors and configured DC power.
+
+    store_estimate writes the separate capacity_estimate config field in memory.
+    Persistence follows the regular EOS configuration save mechanism.
+    The active capacity_wh and raw measurements are never changed here.
+    """
+    try:
+        estimate = await get_measurement().estimate_battery_capacity(battery_id, request)
+        if request.store_estimate:
+            batteries = [b for b in (get_config().devices.batteries or {}).values() if b.device_id == battery_id]
+            if len(batteries) != 1:
+                raise ValueError("Battery configuration changed during estimation; retry.")
+            get_config().set_nested_value(
+                f"devices/batteries/{battery_id}/capacity_estimate", estimate
+            )
+        return estimate
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+
+
+@router.put("/samples")
+async def put_samples(
+    samples: Annotated[list[MeasurementSample], Body(max_length=10000)],
+) -> dict[str, int]:
+    """Upsert raw values and their quality; legacy value/series payloads remain valid."""
+    try:
+        await get_measurement().import_samples(samples)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+    return {"updated": len(samples)}
+
+
+@router.get("/samples", response_model=list[MeasurementSample])
+async def get_samples(key: str, start: AwareDatetime, end: AwareDatetime) -> list[MeasurementSample]:
+    """Read raw samples including quality, in a bounded half-open range."""
+    measurement = get_measurement()
+    try:
+        measurement._energy_channel(key)
+        if not 0 < end.timestamp() - start.timestamp() <= 31 * 86400:
+            raise ValueError("Require a positive range of at most 31 days.")
+        result = []
+        async for record in measurement.db_iterate_records(
+            DatabaseTimestamp.from_datetime(to_datetime(start)),
+            DatabaseTimestamp.from_datetime(to_datetime(end)),
+        ):
+            if key in record.configured_data and record.date_time is not None:
+                result.append(
+                    MeasurementSample(
+                        date_time=record.date_time,
+                        key=key,
+                        value=record.configured_data[key],
+                        quality=record.sample_quality.get(key, SampleQuality()),
+                    )
+                )
+                if len(result) > 10000:
+                    raise ValueError("More than 10000 samples; request a shorter range.")
+        return result
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+
+
+@router.get("/energy", response_model=list[EnergyInterval])
+async def get_energy(
+    key: str,
+    start: AwareDatetime,
+    end: AwareDatetime,
+    interval_seconds: Annotated[int, Query(gt=0)] = 900,
+) -> list[EnergyInterval]:
+    """Energy in Wh, with temporal coverage and quality; no missing-to-zero filling."""
+    try:
+        return await get_measurement().energy_intervals(key, start, end, interval_seconds)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+
+
+@router.get("/household", response_model=dict[str, list[EnergyInterval]])
+async def get_household(
+    start: AwareDatetime, end: AwareDatetime, interval_seconds: Annotated[int, Query(gt=0)] = 900
+) -> dict[str, list[EnergyInterval]]:
+    """Site, household without EV, and base without separately measured devices."""
+    try:
+        return await get_measurement().household_intervals(start, end, interval_seconds)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
diff --git a/tests/test_battery_capacity.py b/tests/test_battery_capacity.py
new file mode 100644
index 00000000..e0ba2666
--- /dev/null
+++ b/tests/test_battery_capacity.py
@@ -0,0 +1,237 @@
+import pytest_asyncio
+"""Capacity fits must preserve energy direction, coverage and independent anchors."""
+
+# ruff: noqa: S101
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from pydantic import ValidationError
+
+from akkudoktoreos.measurement.batterycapacity import (
+    BatteryCapacityEstimationSettings,
+    BatteryCapacityRequest,
+    estimate_capacity,
+)
+from akkudoktoreos.measurement.measurement import MeasurementChannelSettings
+from akkudoktoreos.measurement.quality import SampleQuality
+
+START = datetime(2026, 9, 10, tzinfo=timezone.utc)
+
+
+def fit(
+    points,
+    *,
+    start_soc=20,
+    end_soc=100,
+    efficiency=1,
+    method="hold",
+    polarity="charging",
+    unit="W",
+    end_seconds=3600,
+    quality=None,
+    max_gap=3600,
+):
+    request = BatteryCapacityRequest(
+        start=START,
+        end=START + timedelta(seconds=end_seconds),
+        start_soc_percentage=start_soc,
+        end_soc_percentage=end_soc,
+        soc_reference="voltage_current_anchor",
+    )
+    settings = BatteryCapacityEstimationSettings(power_key="dc", positive_power=polarity)
+    channel = MeasurementChannelSettings(
+        quantity="power",
+        unit=unit,
+        integration_method=method,
+        max_gap_seconds=max_gap,
+    )
+    return estimate_capacity(
+        request,
+        settings,
+        channel,
+        [
+            (START + timedelta(seconds=t), v, (quality or {}).get(t, SampleQuality()))
+            for t, v in points
+        ],
+        battery_id="battery",
+        capacity_wh=12000,
+        charging_efficiency=efficiency,
+        discharging_efficiency=efficiency,
+    )
+
+
+def test_charge_fit_and_unclipped_model_error():
+    result = fit([(0, 8000), (3600, 8000)])
+    assert result.estimated_capacity_wh == pytest.approx(10000)
+    assert result.configured_capacity_wh == 12000
+    assert result.model_end_soc_percentage_unclipped == pytest.approx(86.6666667)
+    assert result.model_soc_error_percentage_points == pytest.approx(-13.3333333)
+    assert result.charge_energy_wh == 8000
+    assert result.coverage_seconds == 3600
+
+
+def test_efficiency_is_applied_once_on_dc_boundary():
+    result = fit([(0, 10000), (3600, 10000)], efficiency=0.8)
+    assert result.estimated_capacity_wh == pytest.approx(10000)
+    assert result.stored_energy_change_wh == 8000
+
+
+def test_discharge_fit_and_reversed_sensor_sign():
+    result = fit(
+        [(0, 6400), (3600, 6400)], start_soc=100, end_soc=20, polarity="discharging", efficiency=0.8
+    )
+    assert result.discharge_energy_wh == 6400
+    assert result.estimated_capacity_wh == pytest.approx(10000)
+
+
+def test_linear_zero_crossing_is_split_before_losses():
+    result = fit([(0, -4000), (3600, 12000)], method="linear", efficiency=0.8)
+    assert result.charge_energy_wh == pytest.approx(4500)
+    assert result.discharge_energy_wh == pytest.approx(500)
+    assert result.stored_energy_change_wh == pytest.approx(4500 * 0.8 - 500 / 0.8)
+
+
+def test_kw_and_clipped_boundary_interpolation():
+    result = fit([(-3600, 0), (3600, 16)], method="linear", unit="kW", max_gap=7200)
+    assert result.charge_energy_wh == pytest.approx(12000)
+    # Deliberately do not saturate the old model at 100%.
+    assert result.model_end_soc_percentage_unclipped == pytest.approx(120)
+
+
+@pytest.mark.parametrize(
+    "points", [[], [(0, 8000)], [(60, 8000), (3600, 8000)], [(0, 8000), (3500, 8000)]]
+)
+def test_missing_coverage_is_not_extrapolated(points):
+    with pytest.raises(ValueError, match="coverage"):
+        fit(points)
+
+
+def test_gap_is_rejected():
+    with pytest.raises(ValueError, match="gap"):
+        fit([(0, 8000), (3600, 8000)], max_gap=300)
+
+
+@pytest.mark.parametrize("value", [None, float("nan"), float("inf"), True])
+def test_bad_power_is_rejected(value):
+    with pytest.raises(ValueError, match="finite measured"):
+        fit([(0, value), (3600, 8000)])
+
+
+@pytest.mark.parametrize("status", ["estimated", "invalid", "unavailable"])
+def test_nonmeasured_quality_is_rejected(status):
+    with pytest.raises(ValueError, match="finite measured"):
+        fit([(0, 8000), (3600, 8000)], quality={0: SampleQuality(status=status)})
+
+
+def test_reset_and_sensor_change_are_rejected():
+    for quality in (SampleQuality(reset=True), SampleQuality(generation="replacement")):
+        with pytest.raises(ValueError, match="reset or generation"):
+            fit([(0, 8000), (3600, 8000)], quality={3600: quality})
+
+
+@pytest.mark.parametrize("start_soc", [100, 99, 85])
+def test_full_to_full_and_small_soc_span_do_not_produce_estimates(start_soc):
+    with pytest.raises(ValueError, match="SoC change is too small"):
+        fit([(0, 8000), (3600, 8000)], start_soc=start_soc)
+
+
+def test_wrong_polarity_is_rejected():
+    with pytest.raises(ValueError, match="disagrees"):
+        fit([(0, -8000), (3600, -8000)])
+
+
+def test_hidden_saturation_cannot_be_fixed_by_end_point_fitting():
+    with pytest.raises(ValueError, match="Fitted SoC leaves"):
+        fit([(0, 24000), (1800, -8000), (3600, -8000)])
+
+
+def test_model_soc_is_not_an_accepted_reference():
+    with pytest.raises(ValidationError):
+        BatteryCapacityRequest(
+            start=START,
+            end=START + timedelta(hours=1),
+            start_soc_percentage=20,
+            soc_reference="calculated_soc",
+        )
+
+
+@pytest_asyncio.fixture
+async def database_case(config_eos):
+    from akkudoktoreos.core.coreabc import get_measurement
+    from akkudoktoreos.measurement.quality import MeasurementSample
+
+    config_eos.merge_settings_from_dict(
+        {
+            "devices": {
+                "batteries": {
+                    "battery": {
+                        "device_id": "battery",
+                        "capacity_wh": 12000,
+                        "charging_efficiency": 1,
+                        "discharging_efficiency": 1,
+                        "capacity_estimation": {
+                            "power_key": "battery_dc",
+                            "positive_power": "charging",
+                        },
+                    }
+                }
+            },
+            "measurement": {
+                "channels": {
+                    "battery_dc": {
+                        "quantity": "power",
+                        "unit": "W",
+                        "integration_method": "hold",
+                        "max_gap_seconds": 3600,
+                    }
+                }
+            },
+        }
+    )
+    (await get_measurement().import_samples(
+        [
+            MeasurementSample(
+                date_time=START + timedelta(seconds=t), key="battery_dc", value=8000.0
+            )
+            for t in (0, 1800, 3600)
+        ]
+    ))
+    return config_eos
+
+
+def test_http_reads_database_and_stores_only_explicit_estimate(database_case):
+    from fastapi.testclient import TestClient
+    from akkudoktoreos.server.eos import app
+
+    client = TestClient(app)
+    body = {
+        "start": START.isoformat(),
+        "end": (START + timedelta(hours=1)).isoformat(),
+        "start_soc_percentage": 20,
+        "soc_reference": "voltage_current_anchor",
+    }
+    response = client.post("/v1/measurement/battery-capacity/battery", json=body)
+    assert response.status_code == 200, response.text
+    assert response.json()["estimated_capacity_wh"] == pytest.approx(10000)
+    battery = database_case.devices.batteries["battery"]
+    assert battery.capacity_estimate is None
+    assert battery.capacity_wh == 12000
+    body["store_estimate"] = True
+    response = client.post("/v1/measurement/battery-capacity/battery", json=body)
+    assert response.status_code == 200, response.text
+    assert battery.capacity_estimate.estimated_capacity_wh == pytest.approx(10000)
+    assert battery.capacity_wh == 12000
+    database_case.merge_settings_from_dict({"optimization": {"genetic": {"individuals": 100}}})
+    battery = database_case.devices.batteries["battery"]
+    assert battery.capacity_estimate.estimated_capacity_wh == pytest.approx(10000)
+    assert battery.capacity_wh == 12000
+
+    # The estimate survives normal config serialization without becoming capacity_wh.
+    data = database_case.to_config_json()
+    assert '"estimated_capacity_wh": 10000.0' in data
+    assert '"capacity_wh": 12000' in data
+    previous = battery.capacity_estimate
+    body["start_soc_percentage"] = 100
+    assert client.post("/v1/measurement/battery-capacity/battery", json=body).status_code == 422
+    assert battery.capacity_estimate is previous
diff --git a/tests/test_consolidation_file_restore.py b/tests/test_consolidation_file_restore.py
new file mode 100644
index 00000000..461d4b04
--- /dev/null
+++ b/tests/test_consolidation_file_restore.py
@@ -0,0 +1,24 @@
+"""Verify fallback JSON loading does not lose records through the singleton."""
+from unittest.mock import AsyncMock
+import pytest
+from akkudoktoreos.core.coreabc import get_measurement
+from akkudoktoreos.core.dataabc import DataSequence
+
+@pytest.mark.asyncio
+async def test_measurement_json_roundtrip(config_eos, tmp_path, monkeypatch):
+    m = get_measurement()
+    config_eos.measurement.load_emr_keys = ["meter"]
+    config_eos.general.data_folder_path = tmp_path
+    config_eos.database.provider = None
+    m._db_reset_state()
+    try:
+        await m.update_value("2026-09-16T08:00:00Z", "meter", 123.5)
+        monkeypatch.setattr(DataSequence, "save", AsyncMock(return_value=False))
+        monkeypatch.setattr(DataSequence, "load", AsyncMock(return_value=False))
+        assert await m.save()
+        m._db_reset_state()
+        assert await m.load()
+        assert len(m.records) == 1
+        assert m.records[0]["meter"] == 123.5
+    finally:
+        m._db_reset_state()
diff --git a/tests/test_measurement_channels.py b/tests/test_measurement_channels.py
new file mode 100644
index 00000000..6167ee32
--- /dev/null
+++ b/tests/test_measurement_channels.py
@@ -0,0 +1,105 @@
+from unittest.mock import AsyncMock
+"""Contracts for typed channels sharing the existing measurement storage."""
+
+# ruff: noqa: S101
+
+import pytest
+from pydantic import ValidationError
+
+from akkudoktoreos.core.coreabc import get_measurement
+from akkudoktoreos.measurement.measurement import (
+    MeasurementChannelSettings,
+    MeasurementCommonSettings,
+    MeasurementDataRecord,
+)
+
+
+CHANNELS = {
+    "house_power": dict(quantity="power", unit="W", integration_method="hold", max_gap_seconds=120),
+    "house_meter": dict(quantity="cumulative_energy", unit="kWh"),
+    "house_interval": dict(
+        quantity="interval_energy", unit="Wh", interval_seconds=900, timestamp_reference="start"
+    ),
+}
+
+
+@pytest.mark.parametrize(
+    "definition",
+    [
+        dict(quantity="power", unit="kWh", integration_method="hold", max_gap_seconds=60),
+        dict(quantity="power", unit="W"),
+        dict(quantity="power", unit="W", integration_method="hold", max_gap_seconds=0),
+        dict(quantity="cumulative_energy", unit="W"),
+        dict(quantity="cumulative_energy", unit="kWh", timestamp_reference="end"),
+        dict(quantity="interval_energy", unit="Wh", interval_seconds=900),
+        dict(
+            quantity="interval_energy", unit="Wh", interval_seconds=True, timestamp_reference="end"
+        ),
+    ],
+)
+def test_reject_ambiguous_channel(definition):
+    with pytest.raises(ValidationError):
+        MeasurementChannelSettings(**definition)
+
+
+@pytest.mark.parametrize(
+    "legacy_field",
+    ["load_emr_keys", "grid_import_emr_keys", "grid_export_emr_keys", "pv_production_emr_keys"],
+)
+def test_legacy_keys_keep_meter_semantics(legacy_field):
+    settings = MeasurementCommonSettings(**{legacy_field: ["legacy"]}, channels=CHANNELS)
+    assert settings.keys == sorted(["legacy", *CHANNELS])
+    compatible = MeasurementCommonSettings(
+        **{legacy_field: ["legacy"]}, channels={"legacy": CHANNELS["house_meter"]}
+    )
+    assert compatible.keys == ["legacy"]
+    with pytest.raises(ValidationError, match="must remain"):
+        MeasurementCommonSettings(
+            **{legacy_field: ["legacy"]}, channels={"legacy": CHANNELS["house_power"]}
+        )
+
+
+@pytest.mark.parametrize("key", ["", " x", "date_time", "configured_data", "keys", "_private"])
+def test_reject_reserved_key(key):
+    with pytest.raises(ValidationError):
+        MeasurementCommonSettings(channels={key: CHANNELS["house_meter"]})
+
+
+@pytest.mark.asyncio
+async def test_existing_import_and_file_reload(config_eos, tmp_path, monkeypatch):
+    """All three quantities retain their raw values through the existing file path."""
+    from akkudoktoreos.core.dataabc import DataSequence
+
+    measurement = get_measurement()
+    previous_settings = config_eos.measurement
+    previous_records = measurement.records
+    previous_folder = config_eos.general.data_folder_path
+    try:
+        config_eos.measurement = MeasurementCommonSettings(channels=CHANNELS)
+        config_eos.general.data_folder_path = tmp_path
+        measurement._db_reset_state()
+        values = dict(house_power=800.0, house_meter=12345.6, house_interval=200.0)
+        for key, value in values.items():
+            (await measurement.update_value("2026-09-10T18:00:00Z", key, value))
+        assert set(values).issubset(measurement.record_keys)
+        monkeypatch.setattr(DataSequence, "save", AsyncMock(return_value=False))
+        monkeypatch.setattr(DataSequence, "load", AsyncMock(return_value=False))
+        assert (await measurement.save())
+        measurement._db_reset_state()
+        assert (await measurement.load())
+        assert len(measurement.records) == 1
+        for key, value in values.items():
+            assert measurement.records[0][key] == value
+        restored = MeasurementDataRecord.model_validate_json(
+            measurement.records[0].model_dump_json()
+        )
+        assert restored.configured_data == values
+        settings = MeasurementCommonSettings.model_validate_json(
+            config_eos.measurement.model_dump_json()
+        )
+        assert settings.model_dump() == config_eos.measurement.model_dump()
+    finally:
+        measurement._db_reset_state()
+        measurement.records = previous_records
+        config_eos.measurement = previous_settings
+        config_eos.general.data_folder_path = previous_folder
diff --git a/tests/test_measurement_energy.py b/tests/test_measurement_energy.py
new file mode 100644
index 00000000..cc15c2b7
--- /dev/null
+++ b/tests/test_measurement_energy.py
@@ -0,0 +1,132 @@
+"""Physical and temporal contracts for interval energy."""
+
+# ruff: noqa: S101
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from zoneinfo import ZoneInfo
+
+from akkudoktoreos.measurement.energy import energy_intervals
+from akkudoktoreos.measurement.measurement import MeasurementChannelSettings
+
+
+START = datetime(2026, 9, 10, tzinfo=timezone.utc)
+
+
+def channel(quantity="power", **kwargs):
+    defaults = {
+        "power": dict(unit="W", integration_method="hold", max_gap_seconds=900),
+        "cumulative_energy": dict(unit="kWh"),
+        "interval_energy": dict(unit="Wh", interval_seconds=900, timestamp_reference="start"),
+    }
+    return MeasurementChannelSettings(quantity=quantity, **(defaults[quantity] | kwargs))
+
+
+def convert(points, config=None, seconds=900):
+    return energy_intervals(
+        [(START + timedelta(seconds=t), v) for t, v in points],
+        config or channel(),
+        START,
+        START + timedelta(seconds=seconds),
+    )
+
+
+@pytest.mark.parametrize(
+    "config,points",
+    [
+        (channel(), [(0, 800), (900, 800)]),
+        (channel(unit="kW"), [(0, 0.8), (900, 0.8)]),
+        (channel("cumulative_energy"), [(0, 10), (900, 10.2)]),
+        (channel("interval_energy"), [(0, 200)]),
+        (channel("interval_energy", timestamp_reference="end"), [(900, 200)]),
+    ],
+)
+def test_equivalent_measurements(config, points):
+    result = convert(points, config)[0]
+    assert result.energy_wh == pytest.approx(200)
+    assert result.coverage_seconds == 900
+    assert result.coverage_status == "complete"
+
+
+def test_time_weighting_and_linear_interpolation():
+    assert convert([(0, 0), (600, 1200), (900, 1200)])[0].energy_wh == 100
+    assert convert([(0, 0), (900, 1600)], channel(integration_method="linear"))[0].energy_wh == 200
+
+
+def test_gap_and_no_extrapolation():
+    result = convert([(0, 800), (300, 800)])[0]
+    assert result.energy_wh is None
+    assert result.observed_energy_wh == pytest.approx(800 / 12)
+    assert result.coverage_status == "partial"
+    assert convert([(0, 800), (900, 800)], channel(max_gap_seconds=60))[0].energy_wh is None
+    assert convert([])[0].coverage_status == "missing"
+
+
+def test_null_breaks_hold_at_outage():
+    result = convert([(0, 800), (300, None), (600, 800), (900, 800)])[0]
+    assert result.coverage_seconds == 600
+    assert result.coverage_status == "partial"
+    assert len(result.coverage_ranges) == 2
+
+
+def test_reset_does_not_create_negative_consumption():
+    result = convert([(0, 10), (450, 0), (900, 0.1)], channel("cumulative_energy"))[0]
+    assert result.energy_wh is None
+    assert result.observed_energy_wh == pytest.approx(100)
+    assert "meter_reset" in result.flags
+
+
+def test_hour_allocation_conserves_energy_and_is_labelled():
+    result = convert([(0, 1000)], channel("interval_energy", interval_seconds=3600), 3600)
+    assert sum(r.energy_wh for r in result) == 1000
+    assert all("allocated_energy" in r.methods for r in result)
+
+
+@pytest.mark.parametrize(
+    "quantity,points",
+    [
+        ("power", [(0, 1), (0, 2)]),
+        ("interval_energy", [(0, 100), (450, 100)]),
+    ],
+)
+def test_ambiguous_time_support_rejected(quantity, points):
+    with pytest.raises(ValueError):
+        convert(points, channel(quantity))
+
+
+def test_nan_and_signed_power():
+    assert convert([(0, float("nan")), (900, 800)])[0].energy_wh is None
+    assert convert([(0, -800), (900, -800)])[0].energy_wh == -200
+
+
+@pytest.mark.parametrize("month,day,hours", [(3, 29, 23), (10, 25, 25)])
+def test_dst_calendar_day(month, day, hours):
+    start = datetime(2026, month, day, tzinfo=ZoneInfo("Europe/Berlin"))
+    end = start + timedelta(days=1)
+    result = energy_intervals([(start, 0), (end, hours)], channel("cumulative_energy"), start, end)
+    assert len(result) == hours * 4
+    assert sum(r.energy_wh for r in result) == pytest.approx(hours * 1000)
+
+
+@pytest.mark.asyncio
+async def test_measurement_wrapper_preserves_asynchronous_channels(config_eos):
+    from akkudoktoreos.core.coreabc import get_measurement
+    from akkudoktoreos.measurement.measurement import MeasurementCommonSettings
+
+    measurement = get_measurement()
+    previous, records = config_eos.measurement, measurement.records
+    try:
+        config_eos.measurement = MeasurementCommonSettings(
+            channels={"p": channel(), "other": channel()}
+        )
+        measurement._db_reset_state()
+        (await measurement.update_value(START, "p", 800))
+        (await measurement.update_value(START + timedelta(seconds=450), "other", 1))
+        (await measurement.update_value(START + timedelta(seconds=900), "p", 800))
+        result = (await measurement.energy_intervals("p", START, START + timedelta(seconds=900)))
+        assert result[0].energy_wh == 200
+    finally:
+        measurement._db_reset_state()
+        measurement.records = records
+        config_eos.measurement = previous
diff --git a/tests/test_measurement_household.py b/tests/test_measurement_household.py
new file mode 100644
index 00000000..525652e4
--- /dev/null
+++ b/tests/test_measurement_household.py
@@ -0,0 +1,366 @@
+"""API, persistence and physical balance contracts for typed measurements."""
+
+# ruff: noqa: S101
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from pydantic import ValidationError
+
+from akkudoktoreos.core.coreabc import get_database, get_measurement
+from akkudoktoreos.measurement.measurement import MeasurementCommonSettings
+from akkudoktoreos.measurement.quality import MeasurementSample
+from akkudoktoreos.server.rest.measurement import router
+
+START = datetime(2026, 9, 10, tzinfo=timezone.utc)
+END = START + timedelta(seconds=900)
+POWER = dict(quantity="power", unit="W", integration_method="hold", max_gap_seconds=900)
+
+
+@pytest.fixture
+def setup_measurement(config_eos):
+    measurement = get_measurement()
+    settings, records = config_eos.measurement, measurement.records
+    measurement._db_reset_state()
+
+    def configure(topology="direct", inputs=None, channels=None):
+        inputs = inputs or [dict(key="site", branch="house", role="site")]
+        config_eos.measurement = MeasurementCommonSettings(
+            channels=channels or {item["key"]: POWER for item in inputs},
+            household=dict(topology=topology, inputs=inputs),
+        )
+        return measurement
+
+    yield configure
+    measurement._db_reset_state()
+    measurement.records = records
+    config_eos.measurement = settings
+
+
+async def write(measurement, key, points):
+    (await measurement.import_samples(
+        [
+            MeasurementSample(
+                date_time=START + timedelta(seconds=t),
+                key=key,
+                value=value,
+                quality=quality if quality else {},
+            )
+            for t, value, quality in points
+        ]
+    ))
+
+
+async def constant(measurement, key, watts):
+    (await write(measurement, key, [(0, watts, None), (900, watts, None)]))
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "topology,values",
+    [
+        ("direct", {"site": 4000}),
+        ("separate_ac", {"grid": 1000, "pv": 2000, "battery": 1000}),
+        ("hybrid_ac", {"grid": 1000, "inverter": 3000}),
+    ],
+)
+async def test_topologies_and_device_subtraction(setup_measurement, topology, values):
+    values = values | {"ev": 2000, "device": 800}
+    m = setup_measurement(topology, [dict(key=k, branch=k, role=k) for k in values])
+    for key, value in values.items():
+        (await constant(m, key, value))
+    rows = (await m.household_intervals(START, END))
+    assert rows["site"][0].energy_wh == 1000
+    assert rows["household"][0].energy_wh == 500
+    assert rows["base"][0].energy_wh == 300
+
+
+@pytest.mark.asyncio
+async def test_export_charge_and_polarity(setup_measurement):
+    m = setup_measurement(
+        "separate_ac",
+        [
+            dict(key="grid", branch="grid", role="grid"),
+            dict(key="solar", branch="solar", role="pv"),
+            dict(key="charge", branch="battery", role="battery", polarity=-1),
+        ],
+    )
+    for key, watts in dict(grid=-1000, solar=4000, charge=1000).items():
+        (await constant(m, key, watts))
+    assert (await m.household_intervals(START, END))["site"][0].energy_wh == 500
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("grid,inverter", [(-6000, 7200), (4200, -3000)])
+async def test_hybrid_with_independent_ac_pv_and_ev(setup_measurement, grid, inverter):
+    inputs = [dict(key=k, branch=k, role=k) for k in ("grid", "inverter", "pv", "ev")]
+    m = setup_measurement("hybrid_ac", inputs)
+    for key, watts in dict(grid=grid, inverter=inverter, pv=400, ev=600).items():
+        (await constant(m, key, watts))
+    rows = (await m.household_intervals(START, END))
+    assert rows["site"][0].energy_wh == 400
+    assert rows["household"][0].energy_wh == 250
+
+
+@pytest.mark.asyncio
+async def test_hybrid_missing_ac_pv_is_not_zero(setup_measurement):
+    m = setup_measurement(
+        "hybrid_ac", [dict(key=k, branch=k, role=k) for k in ("grid", "inverter", "pv")]
+    )
+    (await constant(m, "grid", -6000))
+    (await constant(m, "inverter", 7200))
+    row = (await m.household_intervals(START, END))["site"][0]
+    assert row.energy_wh is None
+    assert row.coverage_seconds == 0
+
+
+def test_hybrid_rejects_separate_battery_to_avoid_double_counting(setup_measurement):
+    with pytest.raises(ValidationError):
+        setup_measurement(
+            "hybrid_ac",
+            [dict(key=k, branch=k, role=k) for k in ("grid", "inverter", "battery")],
+        )
+
+
+@pytest.mark.asyncio
+async def test_missing_ev_does_not_destroy_site(setup_measurement):
+    m = setup_measurement(inputs=[dict(key=k, branch=k, role=k) for k in ("site", "ev")])
+    (await constant(m, "site", 800))
+    rows = (await m.household_intervals(START, END))
+    assert rows["site"][0].energy_wh == 200
+    assert rows["household"][0].energy_wh is None
+    assert rows["household"][0].coverage_seconds == 0
+
+
+@pytest.mark.asyncio
+async def test_intersection_integrates_actual_shape(setup_measurement):
+    m = setup_measurement(inputs=[dict(key=k, branch=k, role=k) for k in ("site", "ev")])
+    (await write(m, "site", [(0, 1000, None), (300, 2000, None), (600, None, None)]))
+    (await write(m, "ev", [(300, 500, None), (900, 500, None)]))
+    row = (await m.household_intervals(START, END))["household"][0]
+    assert row.coverage_seconds == 300
+    assert row.energy_wh is None
+    assert row.observed_energy_wh == pytest.approx(125)
+
+
+@pytest.mark.asyncio
+async def test_nonoverlapping_coverage_is_missing(setup_measurement):
+    m = setup_measurement(inputs=[dict(key=k, branch=k, role=k) for k in ("site", "ev")])
+    (await write(m, "site", [(0, 1000, None), (300, 1000, None)]))
+    (await write(m, "ev", [(600, 500, None), (900, 500, None)]))
+    row = (await m.household_intervals(START, END))["household"][0]
+    assert row.observed_energy_wh is None
+    assert row.coverage_seconds == 0
+
+
+@pytest.mark.parametrize(
+    "inputs",
+    [
+        [dict(key="p", branch="x", role="grid"), dict(key="p", branch="y", role="pv")],
+        [dict(key="p", branch="x", role="grid"), dict(key="q", branch="x", role="pv")],
+        [dict(key="p", branch="x", role="site"), dict(key="q", branch="y", role="pv")],
+        [dict(key="unknown", branch="x", role="grid")],
+    ],
+)
+def test_invalid_balance_configuration(setup_measurement, inputs):
+    with pytest.raises(ValidationError):
+        setup_measurement("separate_ac", inputs, {"p": POWER, "q": POWER})
+
+
+@pytest.mark.asyncio
+async def test_quality_reset_even_with_increasing_meter(setup_measurement):
+    m = setup_measurement(channels={"site": dict(quantity="cumulative_energy", unit="Wh")})
+    (await write(
+        m,
+        "site",
+        [
+            (0, 0, {"generation": "old"}),
+            (450, 1000, {"generation": "new", "reset": True}),
+            (900, 1100, {"generation": "new", "status": "estimated"}),
+        ],
+    ))
+    row = (await m.energy_intervals("site", START, END))[0]
+    assert row.energy_wh is None
+    assert row.observed_energy_wh == 100
+    assert set(row.flags) == {"meter_reset", "estimated"}
+
+
+@pytest.mark.asyncio
+async def test_api_upsert_outage_validation_and_readback(setup_measurement):
+    m = setup_measurement()
+    app = FastAPI()
+    app.include_router(router)
+    with TestClient(app) as client:
+        payload = [
+            dict(
+                date_time=START.isoformat(), key="site", value=800, quality={"status": "estimated"}
+            ),
+            dict(date_time=END.isoformat(), key="site", value=800),
+        ]
+        assert client.put("/v1/measurement/samples", json=payload).status_code == 200
+        params = dict(key="site", start=START.isoformat(), end=END.isoformat())
+        row = client.get("/v1/measurement/energy", params=params).json()[0]
+        assert row["energy_wh"] == 200
+        assert "estimated" in row["flags"]
+        raw = client.get("/v1/measurement/samples", params=params).json()
+        assert raw[0]["quality"]["status"] == "estimated"
+        payload[0]["value"] = None
+        payload[0]["quality"] = {"status": "unavailable"}
+        assert client.put("/v1/measurement/samples", json=payload[:1]).status_code == 200
+        row = client.get("/v1/measurement/energy", params=params).json()[0]
+        assert row["energy_wh"] is None
+        assert "unavailable" in row["flags"]
+        assert len(m.records) == 2
+        payload[0]["value"] = 0
+        payload[0]["quality"] = {}
+        assert client.put("/v1/measurement/samples", json=payload[:1]).status_code == 200
+        assert client.get("/v1/measurement/energy", params=params).json()[0]["energy_wh"] == 0
+        assert (
+            client.get("/v1/measurement/household", params=params).json()["site"][0]["energy_wh"]
+            == 0
+        )
+        # Validate a whole batch before changing any value.
+        bad = [payload[0] | {"value": 999}, payload[0] | {"key": "unknown"}]
+        assert client.put("/v1/measurement/samples", json=bad).status_code == 422
+        assert (await m.energy_intervals("site", START, END))[0].energy_wh == 0
+        assert (
+            client.get(
+                "/v1/measurement/energy", params=params | {"interval_seconds": 0}
+            ).status_code
+            == 422
+        )
+        assert (
+            client.get(
+                "/v1/measurement/energy", params=params | {"start": "2026-09-10T00:00:00"}
+            ).status_code
+            == 422
+        )
+        assert (
+            client.get(
+                "/v1/measurement/energy",
+                params=params | {"end": (START + timedelta(days=32)).isoformat()},
+            ).status_code
+            == 422
+        )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("provider", [None, "SQLite", "LMDB"])
+async def test_quality_persists_through_storage_restart(
+    setup_measurement, config_eos, tmp_path, provider
+):
+    m = setup_measurement()
+    folder, previous_db = config_eos.general.data_folder_path, config_eos.database.provider
+    try:
+        config_eos.general.data_folder_path = tmp_path
+        config_eos.database.provider = provider
+        (await write(
+            m,
+            "site",
+            [
+                (0, 800, {"status": "estimated", "generation": "meter1"}),
+                (450, None, {"status": "unavailable"}),
+                (900, 800, None),
+            ],
+        ))
+        before = (await m.energy_intervals("site", START, END))
+        assert (await m.save())
+        if provider:
+            (await get_database().close())
+        m._db_reset_state()
+        if not provider:
+            assert (await m.load())
+        # DB loads on demand via the bounded query, without a full-history load.
+        assert (await m.energy_intervals("site", START, END)) == before
+        assert m.records[0].sample_quality["site"].generation == "meter1"
+    finally:
+        if provider:
+            (await get_database().close())
+        m._db_reset_state()
+        config_eos.database.provider = previous_db
+        config_eos.general.data_folder_path = folder
+
+
+@pytest.mark.asyncio
+async def test_query_passes_bounded_storage_window(setup_measurement, monkeypatch):
+    m = setup_measurement()
+    calls = []
+    original = type(m).db_iterate_records
+
+    def record_window(self, start_timestamp=None, end_timestamp=None):
+        calls.append((start_timestamp, end_timestamp))
+        return original(self, start_timestamp, end_timestamp)
+
+    monkeypatch.setattr(type(m), "db_iterate_records", record_window)
+    (await m.energy_intervals("site", START, END))
+    assert calls and all(a is not None and b is not None for a, b in calls)
+
+
+@pytest.mark.asyncio
+async def test_actual_server_routes_and_legacy_value_api(setup_measurement):
+    setup_measurement()
+    from akkudoktoreos.server.eos import app
+
+    # No lifespan: this test must not start schedulers or write application state.
+    client = TestClient(app)
+    try:
+        for time in (START, END):
+            response = client.put(
+                "/v1/measurement/value",
+                params={
+                    "datetime": time.isoformat(),
+                    "key": "site",
+                    "value": 800,
+                },
+            )
+            assert response.status_code == 200, response.text
+        response = client.get(
+            "/v1/measurement/energy",
+            params={
+                "key": "site",
+                "start": START.isoformat(),
+                "end": END.isoformat(),
+            },
+        )
+        assert response.status_code == 200
+        assert response.json()[0]["energy_wh"] == 200
+        assert "sample_quality" not in get_measurement().record_keys
+    finally:
+        client.close()
+
+
+@pytest.mark.asyncio
+async def test_quality_merge_keeps_other_channels(setup_measurement):
+    from akkudoktoreos.measurement.measurement import MeasurementDataRecord
+
+    m = setup_measurement(inputs=[dict(key=k, branch=k, role=k) for k in ("site", "ev")])
+    (await write(m, "site", [(0, 800, {"status": "estimated"})]))
+    (await write(m, "ev", [(0, None, {"status": "unavailable"})]))
+    (await m.insert_by_datetime(
+        MeasurementDataRecord(date_time=START, sample_quality={"site": {"status": "measured"}})
+    ))
+    assert m.records[0].sample_quality["site"].status == "measured"
+    assert m.records[0].sample_quality["ev"].status == "unavailable"
+
+
+@pytest.mark.asyncio
+async def test_hour_energy_allocation_and_short_last_interval(setup_measurement):
+    m = setup_measurement(
+        channels={
+            "site": dict(
+                quantity="interval_energy",
+                unit="Wh",
+                interval_seconds=3600,
+                timestamp_reference="start",
+            )
+        }
+    )
+    (await write(m, "site", [(0, 1000, None)]))
+    rows = (await m.household_intervals(START, START + timedelta(seconds=1800)))["site"]
+    assert [row.energy_wh for row in rows] == [250, 250]
+    assert all("allocated_energy" in row.methods for row in rows)
+    row = (await m.energy_intervals("site", START, START + timedelta(seconds=60)))[0]
+    assert row.coverage_seconds == 60
+    assert row.energy_wh == pytest.approx(1000 / 60)