feat(pvforecast): add adaptive per-plane PV learning and diagnostics (#1422)

PV arrays with different orientations and curtailment need separate plant calibration and weather correction. This opt-in provider publishes normal EOS AC/DC forecast keys at 15-minute resolution.

- Preserve daily per-plane energy/shape calibration and validated short-term local/trend selection. Curtailment-sensitive arrays require a fresh release flag or an explicitly configured learning guard.
-  Add switchable empirical solar-angle response relative to separate eligible reference arrays, removing their common weather error. This learns angular residuals, not physical angles or a measured geometric horizon.
-  Archive actually received weather model inputs, first local receipt and frozen per-member PV candidates. Returned past weather is excluded; unknown source issuance remains explicit. Learn model weights and forecast-regime bias separately at 2–6, 6–24 and 24–48 hours.
- New long-horizon candidates remain in shadow until their genuinely issued outputs improve equal-day MAE by the configured margin, win on at least 60% of validation days and pass a matched-support energy-bias guard. Insufficient evidence retains calibration. Daily yields still require complete local days, including DST.
- Expose fitted parameters, independent training/validation day counts, errors, selected methods and persistence in the existing status API, with additional normal prediction keys for Grafana. Retain physical DC through angular/weather corrections an  enforce individual inverter AC limits.

Co-authored-by: Normann <github@koldrack.com>
Co-authored-by: Christin <chrissi@von-perbandt.de>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: YMT <yn276364@dal.ca>
Co-authored-by: sarankumar1325 <120080287+sarankumar1325@users.noreply.github.com>

src/akkudoktoreos/core/database.py

@@ -747,7 +747,7 @@ class SQLiteDatabase(DatabaseBackendABC):
 
         ns = self._ns(namespace)
 
-        with self.conn:
+        with self.lock, self.conn:
             # Ensure metadata table exists
             self.conn.execute("""
                 CREATE TABLE IF NOT EXISTS metadata (
@@ -821,6 +821,9 @@ class SQLiteDatabase(DatabaseBackendABC):
 
         with self.lock:
             with self.conn:
+                # isolation_level=None disables implicit transactions. Without
+                # BEGIN each row commits separately and failed batches are partial.
+                self.conn.execute("BEGIN IMMEDIATE")
                 self.conn.executemany(
                     "INSERT OR REPLACE INTO records (namespace, key, value) VALUES (?, ?, ?)",
                     rows,
@@ -850,6 +853,7 @@ class SQLiteDatabase(DatabaseBackendABC):
 
         with self.lock:
             with self.conn:
+                self.conn.execute("BEGIN IMMEDIATE")
                 cursor = self.conn.executemany(
                     "DELETE FROM records WHERE namespace = ? AND key = ?",
                     rows,

src/akkudoktoreos/prediction/prediction.py

@@ -53,6 +53,9 @@ from akkudoktoreos.prediction.loadakkudoktor import (
 from akkudoktoreos.prediction.loadimport import LoadImport
 from akkudoktoreos.prediction.loadvrm import LoadVrm
 from akkudoktoreos.prediction.predictionabc import PredictionContainer
+from akkudoktoreos.prediction.pvakkuadapt import (
+    PVForecastAdaptiveAkkudoktor,
+)
 from akkudoktoreos.prediction.pvforecastakkudoktor import PVForecastAkkudoktor
 from akkudoktoreos.prediction.pvforecastforecastsolar import PVForecastForecastSolar
 from akkudoktoreos.prediction.pvforecasthomeassistant import PVForecastHomeAssistant
@@ -106,6 +109,7 @@ loadforecast_akkudoktor_adjusted = LoadAkkudoktorAdjusted()
 loadforecast_vrm = LoadVrm()
 loadforecast_import = LoadImport()
 pvforecast_akkudoktor = PVForecastAkkudoktor()
+pvforecast_adaptive_akkudoktor = PVForecastAdaptiveAkkudoktor()
 pvforecast_vrm = PVForecastVrm()
 pvforecast_homeassistant = PVForecastHomeAssistant()
 pvforecast_pvlib = PVForecastPVLib()
@@ -140,6 +144,7 @@ PredictionProviderType = Union[
     LoadImport,
     LoadVrm,
     PVForecastAkkudoktor,
+    PVForecastAdaptiveAkkudoktor,
     PVForecastForecastSolar,
     PVForecastHomeAssistant,
     PVForecastImport,
@@ -180,6 +185,7 @@ def prediction_providers() -> list[PredictionProviderType]:
         loadforecast_vrm, \
         loadforecast_import, \
         pvforecast_akkudoktor, \
+        pvforecast_adaptive_akkudoktor, \
         pvforecast_vrm, \
         pvforecast_homeassistant, \
         pvforecast_pvlib, \
@@ -221,6 +227,7 @@ def prediction_providers() -> list[PredictionProviderType]:
         loadforecast_import,
         loadforecast_vrm,
         pvforecast_akkudoktor,
+        pvforecast_adaptive_akkudoktor,
         pvforecast_forecastsolar,
         pvforecast_homeassistant,
         pvforecast_import,

src/akkudoktoreos/prediction/pvakkuadaptmodel.py

@@ -0,0 +1,304 @@
+"""Small causal PV calibration model shared by the provider and offline backtest."""
+
+from dataclasses import dataclass, field
+
+import numpy as np
+import pandas as pd
+
+
+@dataclass
+class AdaptivePVModel:
+    """Learn energy bias and a smooth intraday shape from completed measurements.
+
+    Inputs are actual 15-minute or hourly interval-start timestamps and mean powers in W.
+    ``available_at``
+    records when a measured interval became usable, including any ingestion lag.
+    This is statistical calibration, not identification of physical fault causes.
+    """
+
+    timezone: str = "Europe/Berlin"
+    learning_days: int = 60
+    min_days: int = 4
+    half_life_days: float = 30.0
+    shape_prior_kwh: float = 2.0
+    min_factor: float = 0.2
+    max_factor: float = 3.0
+    interval_minutes: int = 15
+    allow_partial_days: bool = False
+    min_partial_day_samples: int = 8
+    gain: float = field(default=1.0, init=False)
+    shape: np.ndarray = field(default_factory=lambda: np.ones(12), init=False)
+    trained_days: int = field(default=0, init=False)
+
+    def fit(self, history: pd.DataFrame, cutoff: pd.Timestamp) -> "AdaptivePVModel":
+        """Fit only completed previous local days available before ``cutoff``.
+
+        Global energy gain uses complete days. Shape uses valid paired intervals
+        from those days, with shrinkage and recency weighting. Missing values,
+        negative production, and explicitly invalid samples never become zeros.
+        With allow_partial_days, the caller must first censor every interval
+        without positive unrestricted-operation evidence. Suitable days then
+        require min_partial_day_samples daylight pairs rather than night data.
+        """
+        self.gain, self.shape, self.trained_days = 1.0, np.ones(12), 0
+        cutoff = _aware(cutoff)
+        if self.interval_minutes not in (15, 60):
+            raise ValueError("Training intervals must be 15 or 60 minutes.")
+        if history.empty:
+            return self
+        history = history.sort_index()
+        history_index = _aware_index(history.index)
+        utc = history_index.tz_convert("UTC")
+        if not utc.equals(utc.floor(f"{self.interval_minutes}min")):
+            raise ValueError("Training timestamps must align with their actual interval.")
+        if history.index.has_duplicates:
+            raise ValueError("One training pair per interval is required.")
+        available = pd.to_datetime(history["available_at"], utc=True)
+        local_day = history_index.tz_convert(self.timezone).normalize()
+        today = cutoff.tz_convert(self.timezone).normalize()
+        valid = (
+            (available <= cutoff)
+            & (local_day < today)
+            & (history.index >= cutoff - pd.Timedelta(days=self.learning_days))
+            & np.isfinite(history["raw_power"])
+            & np.isfinite(history["measured_power"])
+            & (history["raw_power"] >= 0)
+            & (history["measured_power"] >= 0)
+        )
+        pairs = history.loc[valid, ["raw_power", "measured_power"]].copy()
+        if pairs.empty:
+            return self
+        pairs["day"] = _aware_index(pairs.index).tz_convert(self.timezone).normalize()
+        # Count against elapsed local days, including daylight-saving transitions.
+        days = pairs.groupby("day").agg(
+            raw=("raw_power", "sum"), actual=("measured_power", "sum"), n=("raw_power", "size")
+        )
+        expected = np.array(
+            [
+                (day + pd.DateOffset(days=1) - day).total_seconds() / (self.interval_minutes * 60)
+                for day in days.index
+            ]
+        )
+        days = days[(days["n"].to_numpy() == expected) & (days["raw"] > 200.0)]
+        if self.allow_partial_days:
+            # The caller has positively identified unrestricted intervals.
+            # Unknown/curtailed intervals have already been removed, never zeroed.
+            daylight = pairs[pairs.raw_power > 50].groupby("day").size()
+            days = pairs.groupby("day").agg(
+                raw=("raw_power", "sum"), actual=("measured_power", "sum"), n=("raw_power", "size")
+            )
+            days = days[
+                (daylight.reindex(days.index, fill_value=0) >= self.min_partial_day_samples)
+                & (days.raw > 200)
+            ]
+        self.trained_days = len(days)
+        if self.trained_days < self.min_days:
+            return self
+        pairs = pairs[pairs["day"].isin(days.index)]
+        age_days = np.array([(cutoff - day).total_seconds() / 86400 for day in days.index])
+        day_weights = np.exp2(-age_days / self.half_life_days)
+        self.gain = float(
+            np.clip(
+                np.dot(days["actual"], day_weights) / np.dot(days["raw"], day_weights),
+                self.min_factor,
+                self.max_factor,
+            )
+        )
+        local = _aware_index(pairs.index).tz_convert(self.timezone)
+        hours = local.hour.to_numpy() + local.minute.to_numpy() / 60 + self.interval_minutes / 120
+        weights = np.exp2(
+            (_aware_index(pairs.index) - cutoff).total_seconds().to_numpy()
+            / 86400
+            / self.half_life_days
+        )
+        prior_wh = self.shape_prior_kwh * 1000
+        for number, center in enumerate(np.arange(12) * 2 + 1):
+            distance = np.abs(hours - center)
+            distance = np.minimum(distance, 24 - distance)
+            kernel = np.maximum(0, 1 - distance / 2) * weights
+            raw_wh = float(np.dot(kernel, pairs["raw_power"])) * self.interval_minutes / 60
+            actual_wh = float(np.dot(kernel, pairs["measured_power"])) * self.interval_minutes / 60
+            self.shape[number] = np.clip(
+                (actual_wh + prior_wh * self.gain) / (raw_wh + prior_wh),
+                self.min_factor,
+                self.max_factor,
+            )
+        return self
+
+    def predict(
+        self, raw_power: pd.Series, capacity_w: float = 0.0, resolution_minutes: int = 15
+    ) -> pd.Series:
+        """Correct shape, conserving the independently learned energy per day."""
+        local = _aware_index(raw_power.index).tz_convert(self.timezone)
+        hours = local.hour.to_numpy() + local.minute.to_numpy() / 60 + resolution_minutes / 120
+        centers = np.arange(12) * 2 + 1
+        scales = np.interp(
+            hours,
+            np.r_[centers[-1] - 24, centers, centers[0] + 24],
+            np.r_[self.shape[-1], self.shape, self.shape[0]],
+        )
+        raw = raw_power.to_numpy(dtype=float)
+        days = local.normalize()
+        for day in days.unique():
+            mask = days == day
+            before = float(np.nansum(raw[mask]))
+            after = float(np.nansum(raw[mask] * scales[mask]))
+            if before > 0 and after > 0:
+                # Bounded multiplicative correction with energy constraint. Bisection
+                # avoids a renormalization violating the configured factor limits.
+                target = before * self.gain
+                lo, hi = 0.0, self.max_factor / max(float(scales[mask].min()), 1e-12)
+                for _ in range(40):
+                    multiplier = (lo + hi) / 2
+                    energy = np.nansum(
+                        raw[mask]
+                        * np.clip(scales[mask] * multiplier, self.min_factor, self.max_factor)
+                    )
+                    if energy < target:
+                        lo = multiplier
+                    else:
+                        hi = multiplier
+                scales[mask] = np.clip(
+                    scales[mask] * (lo + hi) / 2, self.min_factor, self.max_factor
+                )
+        corrected = raw_power * scales
+        return corrected.clip(lower=0, upper=capacity_w if capacity_w > 0 else None)
+
+
+def apply_nowcast(
+    prediction: pd.Series,
+    history: pd.DataFrame,
+    cutoff: pd.Timestamp,
+    half_life_hours: float = 1.0,
+    max_age_minutes: int = 30,
+    capacity_w: float = 0.0,
+) -> pd.Series:
+    """Blend a recent measured/modelled ratio back to the forecast with lead time.
+
+    Require at least 30 minutes of actual support in the past hour. The short-term
+    correction does not survive overnight, is clamped, and expires with stale data.
+    History must contain ``calibrated_power``, not already nowcast predictions.
+    """
+    cutoff = _aware(cutoff)
+    if history.empty:
+        return prediction.copy()
+    _aware_index(history.index)
+    available = pd.to_datetime(history["available_at"], utc=True)
+    usable = history[
+        (available <= cutoff)
+        & (history.index + pd.Timedelta(minutes=15) <= cutoff)
+        & (history.index >= cutoff - pd.Timedelta(hours=1))
+    ]
+    usable = usable[
+        np.isfinite(usable["calibrated_power"])
+        & np.isfinite(usable["measured_power"])
+        & (usable["calibrated_power"] > 50)
+        & (usable["measured_power"] >= 0)
+    ]
+    if len(usable) < 2:
+        return prediction.copy()
+    latest_end = usable.index.max() + pd.Timedelta(minutes=15)
+    if cutoff - latest_end > pd.Timedelta(minutes=max_age_minutes):
+        return prediction.copy()
+    ratio = float(
+        np.clip(usable["measured_power"].sum() / usable["calibrated_power"].sum(), 0.5, 1.5)
+    )
+    lead = np.maximum(0, (prediction.index - latest_end).total_seconds().to_numpy() / 3600)
+    scale = 1 + (ratio - 1) * np.exp2(-lead / half_life_hours)
+    same_day = _aware_index(prediction.index).tz_convert(cutoff.tz).date == cutoff.date()
+    scale = np.where((prediction.index >= cutoff) & same_day, scale, 1.0)
+    return (prediction * scale).clip(lower=0, upper=capacity_w if capacity_w > 0 else None)
+
+
+def apply_local_nowcast(
+    prediction: pd.Series,
+    clear_sky_power: pd.Series,
+    history: pd.DataFrame,
+    cutoff: pd.Timestamp,
+    half_life_minutes: float = 30.0,
+    horizon_minutes: int = 120,
+    max_age_minutes: int = 30,
+    capacity_w: float = 0.0,
+    reference_interval_minutes: int = 15,
+) -> pd.Series:
+    """Carry a recent local weather residual toward the current weather forecast.
+
+    Reference measurements may be a non-curtailed subset of the plant. Normalize
+    their residual by their own clear-sky power, then transfer it to the whole
+    plant's clear-sky power. Require two contiguous complete reference intervals.
+    Missing, stale, unavailable or demand-limited references yield no correction.
+    ``eligible`` is optional; callers must exclude curtailment using measurement
+    quality or provide explicit eligibility. This does not infer cloud motion.
+    """
+    cutoff = _aware(cutoff)
+    prediction_index = _aware_index(prediction.index)
+    if reference_interval_minutes not in (5, 15):
+        raise ValueError("Local reference intervals must be 5 or 15 minutes.")
+    _aware_index(history.index, reference_interval_minutes)
+    if half_life_minutes <= 0 or horizon_minutes <= 0:
+        raise ValueError("Local nowcast decay and horizon must be positive.")
+    if history.index.has_duplicates:
+        raise ValueError("One reference measurement per interval is required.")
+    if history.empty:
+        return prediction.copy()
+    available = pd.to_datetime(history["available_at"], utc=True)
+    recent = (
+        history[
+            (available <= cutoff)
+            & (history.index + pd.Timedelta(minutes=reference_interval_minutes) <= cutoff)
+            & (history.index >= cutoff - pd.Timedelta(hours=1))
+        ]
+        .sort_index()
+        .tail(2)
+    )
+    reference_index = _aware_index(recent.index, reference_interval_minutes)
+    if len(recent) != 2 or reference_index[1] - reference_index[0] != pd.Timedelta(
+        minutes=reference_interval_minutes
+    ):
+        return prediction.copy()
+    end = reference_index[-1] + pd.Timedelta(minutes=reference_interval_minutes)
+    if cutoff - end > pd.Timedelta(minutes=max_age_minutes):
+        return prediction.copy()
+    if (
+        not np.isfinite(recent[["calibrated_power", "measured_power", "clear_sky_power"]])
+        .all()
+        .all()
+    ):
+        return prediction.copy()
+    if (recent["measured_power"] < 0).any() or (recent["clear_sky_power"] < 150).any():
+        return prediction.copy()
+    if "eligible" in recent and not recent["eligible"].fillna(False).all():
+        return prediction.copy()
+    residual = (
+        (recent["measured_power"] - recent["calibrated_power"]) / recent["clear_sky_power"]
+    ).clip(-1.5, 1.5)
+    # Fixed recency weighting: no trend extrapolation from a passing cloud.
+    local_error = float(np.dot(residual.to_numpy(), [0.35, 0.65]))
+    age = (
+        cutoff - (reference_index[-1] + pd.Timedelta(minutes=reference_interval_minutes / 2))
+    ).total_seconds() / 60
+    lead = (prediction_index - cutoff).total_seconds().to_numpy() / 60
+    decay = np.exp2(-np.maximum(0, lead + age) / half_life_minutes)
+    # Fade to exactly zero, avoiding a step at the finite horizon.
+    taper = np.clip(2 * (1 - lead / horizon_minutes), 0, 1)
+    same_day = prediction_index.tz_convert(cutoff.tz).date == cutoff.date()
+    usable = (lead >= 0) & (lead < horizon_minutes) & same_day
+    envelope = clear_sky_power.reindex(prediction.index)
+    correction = envelope * np.where(usable, local_error * decay * taper, 0.0)
+    result = (prediction + correction).where(envelope.notna(), prediction)
+    return result.clip(lower=0, upper=capacity_w if capacity_w > 0 else None)
+
+
+def _aware(value: pd.Timestamp) -> pd.Timestamp:
+    if value.tzinfo is None:
+        raise ValueError("Explicit timezone required.")
+    return value
+
+
+def _aware_index(index: pd.Index, resolution_minutes: int = 15) -> pd.DatetimeIndex:
+    if not isinstance(index, pd.DatetimeIndex) or index.tz is None:
+        raise ValueError("Timezone-aware DatetimeIndex required.")
+    utc = index.tz_convert("UTC")
+    if not utc.equals(utc.floor(f"{resolution_minutes}min")):
+        raise ValueError(f"PV intervals must start on {resolution_minutes}-minute boundaries.")
+    return index

src/akkudoktoreos/prediction/pvakkuadaptquality.py

@@ -0,0 +1,148 @@
+"""Causal selection of issued PV candidates on completed, independent target days."""
+
+from typing import Any, cast
+
+import numpy as np
+import pandas as pd
+
+METHODS = ("calibrated", "local", "trend")
+
+
+def select_corrections(
+    rows: list[dict[str, Any]],
+    observations: pd.DataFrame,
+    groups: dict[str, list[str]],
+    cutoff: pd.Timestamp,
+    timezone: str,
+    reference_group: str | None = None,
+    validation_days: int = 7,
+    minimum_days: int = 3,
+    minimum_samples: int = 32,
+    improvement: float = 0.02,
+    enabled_methods: tuple[str, ...] = METHODS,
+) -> dict[str, Any]:
+    """Compare frozen candidates, never coefficients refitted to their own labels.
+
+    Each target interval gets one vote per lead, irrespective of polling rate.
+    Predictions issued today, unavailable labels and censored observations do
+    not validate a candidate. Ties and insufficient evidence retain calibration.
+    Legacy archives can validate only their explicitly recorded reference sum.
+    """
+    today = cutoff.tz_convert(timezone).normalize()
+    start = today - pd.DateOffset(days=validation_days)
+    start_ms, end_ms = int(start.timestamp() * 1000), int(today.timestamp() * 1000)
+    if (
+        not isinstance(observations.index, pd.DatetimeIndex)
+        or observations.index.tz is None
+        or observations.index.has_duplicates
+    ):
+        raise ValueError("Validation requires distinct, timezone-aware observation intervals.")
+    selected: dict[str, Any] = {}
+    for group, keys in groups.items():
+        comparisons = []
+        truth = {}
+        for timestamp, values_row in observations.iterrows():
+            target_time = cast(pd.Timestamp, timestamp)
+            if not start <= target_time < today:
+                continue
+            obs = cast("pd.Series[Any]", values_row)
+            if all(
+                pd.notna(obs.get(key))
+                and pd.notna(obs.get(key + "_eligible", True))
+                and bool(obs.get(key + "_eligible", True))
+                and pd.notna(obs.get(key + "_available"))
+                and obs[key + "_available"] <= cutoff
+                for key in keys
+            ):
+                truth[int(target_time.timestamp() * 1000)] = (
+                    sum(float(obs[key]) for key in keys),
+                    str(target_time.tz_convert(timezone).date()),
+                )
+        for row in rows:
+            target_ms, issued_ms = row["target_ms"], row["issue_ms"]
+            lead = float(row["lead_minutes"])
+            if (
+                not start_ms <= target_ms < end_ms
+                or not issued_ms < target_ms
+                or not 0 <= lead < 120
+            ):
+                continue
+            if target_ms not in truth:
+                continue
+            candidate = row.get("groups", {}).get(group)
+            if candidate is None:
+                if group != reference_group:
+                    continue
+                candidate = {
+                    "calibrated": row.get("reference_calibrated_w"),
+                    "local": row.get("reference_local_w"),
+                    "trend": row.get("reference_adaptive_w"),
+                }
+            values = [candidate.get(method) for method in METHODS]
+            if any(value is None or not np.isfinite(value) for value in values):
+                continue
+            actual, day = truth[target_ms]
+            if actual <= 50 and float(candidate["calibrated"]) <= 50:
+                continue
+            bucket = int(np.ceil(lead / 15) * 15)
+            comparisons.append(
+                dict(
+                    target=row["target_ms"],
+                    bucket=bucket,
+                    distance=abs(lead - bucket + 7.5),
+                    day=day,
+                    **{method: abs(float(candidate[method]) - actual) for method in METHODS},
+                )
+            )
+        per_lead = {}
+        if comparisons:
+            frame = (
+                pd.DataFrame(comparisons)
+                .sort_values("distance")
+                .drop_duplicates(["target", "bucket"])
+            )
+            for bucket_label, part in frame.groupby("bucket"):
+                # Equal day weighting prevents a bright long day or faster poller
+                # from choosing a model for every other day.
+                errors = part.groupby("day")[list(METHODS)].mean().mean()
+                days = int(part.day.nunique())
+                enough = days >= minimum_days and len(part) >= minimum_samples
+                choice = "calibrated"
+                best = float(errors[choice])
+                for method in METHODS[1:]:
+                    value = float(errors[method])
+                    if method in enabled_methods and enough and value < best * (1 - improvement):
+                        choice, best = method, value
+                per_lead[str(int(cast(int, bucket_label)))] = dict(
+                    method=choice,
+                    days=days,
+                    samples=len(part),
+                    validated=enough,
+                    mae_w={method: float(errors[method]) for method in METHODS},
+                )
+        selected[group] = per_lead
+    return selected
+
+
+def choose_correction(
+    calibrated: pd.Series,
+    local: pd.Series,
+    trend: pd.Series,
+    issue: pd.Timestamp,
+    validation: dict[str, Any],
+    mode: str = "auto",
+) -> pd.Series:
+    """Use a validated candidate per lead; outside two hours keep calibration."""
+    result = calibrated.copy()
+    leads = (pd.DatetimeIndex(calibrated.index) - issue).total_seconds().to_numpy() / 60
+    buckets = np.ceil(leads / 15).astype(int) * 15
+    candidates = {"calibrated": calibrated, "local": local, "trend": trend}
+    for bucket in np.unique(buckets):
+        method = mode
+        if mode == "auto":
+            method = validation.get(str(int(bucket)), {}).get("method", "calibrated")
+        if method not in candidates:
+            method = "calibrated"
+        mask = (buckets == bucket) & (leads >= 0) & (leads < 120)
+        result.loc[mask] = candidates[method].loc[mask]
+    return result

src/akkudoktoreos/prediction/pvakkuadaptresponse.py

@@ -0,0 +1,262 @@
+"""Regularized angular response and issued-weather calibration with causal guards."""
+
+from typing import Any, cast
+
+import numpy as np
+import pandas as pd
+
+LEADS = ("2-6h", "6-24h", "24-48h")
+METHODS = ("calibrated", "solar", "weather")
+
+
+def _sum_powers(values: list[Any]) -> float:
+    if any(value is None for value in values):
+        return float("nan")
+    return float(np.sum(values))
+
+
+def lead_bucket(minutes: float) -> str | None:
+    """Separate intraday, next-day and second-day weather skill."""
+    if 120 <= minutes < 360:
+        return LEADS[0]
+    if 360 <= minutes < 1440:
+        return LEADS[1]
+    if 1440 <= minutes <= 2880:
+        return LEADS[2]
+    return None
+
+
+def solar_basis(index: pd.DatetimeIndex, latitude: float, longitude: float) -> np.ndarray:
+    """Smooth local angular basis at interval centers, independent of civil time."""
+    import pvlib
+
+    solar = pvlib.solarposition.get_solarposition(
+        index + pd.Timedelta(minutes=7.5), latitude, longitude
+    )
+    azimuth = solar.azimuth.to_numpy()
+    elevation = solar.apparent_elevation.to_numpy()
+    az = np.exp(-0.5 * ((azimuth[:, None] - np.arange(60, 301, 30)) / 22) ** 2)
+    el = np.exp(-0.5 * ((elevation[:, None] - np.arange(0, 81, 20)) / 14) ** 2)
+    basis = (az[:, :, None] * el[:, None, :]).reshape(len(index), 45)
+    return basis / np.maximum(basis.sum(axis=1, keepdims=True), 1e-9)
+
+
+def fit_solar_response(
+    history: pd.DataFrame,
+    cutoff: pd.Timestamp,
+    timezone: str,
+    latitude: float,
+    longitude: float,
+    minimum_days: int,
+) -> dict[str, Any]:
+    """Learn differential array response after removing common reference weather.
+
+    This estimates repeatable angular residuals, not a measured geometric horizon.
+    Training requires eligible observations and a separate independent reference.
+    Unseen angular regions retain the prior response of one.
+    """
+    result: dict[str, Any] = dict(coefficients=[0.0] * 45, days=0, samples=0, fitted=False)
+    if history.empty:
+        return result
+    today = cutoff.tz_convert(timezone).normalize()
+    valid = (
+        (history.index < today)
+        & (history.available_at <= cutoff)
+        & np.isfinite(history[["base", "actual", "common"]]).all(axis=1)
+        & (history.base > 80)
+        & (history.actual >= 0)
+        & history.common.between(0.35, 2.5)
+    )
+    pairs = history.loc[valid].copy()
+    days = pd.DatetimeIndex(pairs.index).tz_convert(timezone).normalize().nunique()
+    result.update(days=int(days), samples=len(pairs))
+    if days < minimum_days or len(pairs) < 48:
+        return result
+    basis = solar_basis(pd.DatetimeIndex(pairs.index), latitude, longitude)
+    target = (pairs.actual / pairs.base / pairs.common).clip(0.35, 1.8).to_numpy() - 1
+    age = (cutoff - pd.DatetimeIndex(pairs.index)).total_seconds().to_numpy() / 86400
+    weights = np.exp2(-age / 14) * np.minimum(1, pairs.base.to_numpy() / 800)
+    # Downweight outliers; strong identity prior shrinks sparsely covered knots.
+    weights *= np.minimum(1, 0.35 / np.maximum(np.abs(target), 1e-9))
+    matrix = basis.T @ (weights[:, None] * basis) + np.eye(45) * 8
+    coefficients = np.linalg.solve(matrix, basis.T @ (weights * target))
+    result.update(coefficients=np.clip(coefficients, -0.5, 0.5).tolist(), fitted=True)
+    return result
+
+
+def solar_factors(basis: np.ndarray, model: dict[str, Any]) -> np.ndarray:
+    """Bound a fitted empirical response; cold start is exactly identity."""
+    coefficients = np.asarray(model.get("coefficients", [0.0] * 45), dtype=float)
+    if coefficients.shape != (45,) or not np.isfinite(coefficients).all():
+        raise ValueError("Invalid adaptive solar response")
+    return np.clip(1 + basis @ coefficients, 0.5, 1.5)
+
+
+def regime(clearness: float) -> str:
+    """Use only forecast irradiance to identify a weather regime."""
+    return "cloudy" if clearness < 0.3 else "mixed" if clearness < 0.7 else "clear"
+
+
+def paired_rows(
+    rows: list[dict[str, Any]],
+    observations: pd.DataFrame,
+    keys: list[str],
+    cutoff: pd.Timestamp,
+    timezone: str,
+) -> list[dict[str, Any]]:
+    """Join frozen inputs to causally available, positively eligible observations.
+
+    One target gets one vote per horizon, independent of the polling frequency.
+    All candidates and member models are compared on identical support.
+    """
+    today = cutoff.tz_convert(timezone).normalize()
+    paired: dict[tuple[int, str], dict[str, Any]] = {}
+    centers = {"2-6h": 240, "6-24h": 900, "24-48h": 2160}
+    for row in rows:
+        target = pd.Timestamp(row["target_ms"], unit="ms", tz="UTC")
+        issued = pd.Timestamp(row["issued_ms"], unit="ms", tz="UTC")
+        lead = (target - issued).total_seconds() / 60
+        bucket = lead_bucket(lead)
+        if (
+            bucket is None
+            or target >= today
+            or target not in observations.index
+            or row.get("response_version") != 1
+        ):
+            continue
+        truth = cast("pd.Series[Any]", observations.loc[target])
+        if not all(
+            pd.notna(truth.get(key))
+            and pd.notna(truth.get(key + "_eligible"))
+            and bool(truth.get(key + "_eligible", False))
+            and pd.notna(truth.get(key + "_available"))
+            and truth[key + "_available"] <= cutoff
+            for key in keys
+        ):
+            continue
+        planes = row.get("planes", {})
+        if not all(key in planes for key in keys):
+            continue
+        values = {
+            method: _sum_powers([planes[key].get(method, np.nan) for key in keys])
+            for method in METHODS
+        }
+        actual = sum(float(truth[key]) for key in keys)
+        if not np.isfinite(list(values.values())).all() or max(actual, values["calibrated"]) < 50:
+            continue
+        members = {
+            name: _sum_powers([member.get(key, np.nan) for key in keys])
+            for name, member in row.get("members", {}).items()
+        }
+        sky = _sum_powers([planes[key].get("sky", np.nan) for key in keys])
+        record = dict(
+            target_ms=row["target_ms"],
+            bucket=bucket,
+            actual=actual,
+            day=str(target.tz_convert(timezone).date()),
+            members=members,
+            distance=abs(lead - centers[bucket]),
+            sky=sky,
+            **values,
+        )
+        identity = (row["target_ms"], bucket)
+        if identity not in paired or record["distance"] < paired[identity]["distance"]:
+            paired[identity] = record
+    return list(paired.values())
+
+
+def fit_weather_response(
+    pairs: list[dict[str, Any]],
+    minimum_days: int,
+    minimum_samples: int,
+) -> dict[str, Any]:
+    """Rank actually issued model members and shrink regime bias toward identity."""
+    result = {}
+    for bucket in LEADS:
+        part = [row for row in pairs if row["bucket"] == bucket]
+        names = set.intersection(*(set(row["members"]) for row in part)) if part else set()
+        part = [
+            row
+            for row in part
+            if np.isfinite([row["members"][name] for name in names]).all()
+            and np.isfinite(row["sky"])
+            and row["sky"] > 50
+        ]
+        days = len({row["day"] for row in part})
+        value: dict[str, Any] = dict(
+            days=days, samples=len(part), weights={}, regimes={}, fitted=False
+        )
+        if names and days >= minimum_days and len(part) >= minimum_samples:
+            errors = {}
+            for name in sorted(names):
+                frame = pd.DataFrame(
+                    dict(day=row["day"], error=abs(row["members"][name] - row["actual"]))
+                    for row in part
+                )
+                errors[name] = float(frame.groupby("day").error.mean().mean())
+            scores = {name: 1 / max(error, 20) ** 2 for name, error in errors.items()}
+            weights = {
+                name: 0.5 / len(names) + 0.5 * score / sum(scores.values())
+                for name, score in scores.items()
+            }
+            for row in part:
+                prediction = sum(weights[name] * row["members"][name] for name in names)
+                row["weather_prediction"] = prediction
+                row["regime"] = regime(prediction / row["sky"])
+            factors = {}
+            for name in ("cloudy", "mixed", "clear"):
+                subset = [row for row in part if row["regime"] == name]
+                if len({row["day"] for row in subset}) >= minimum_days:
+                    predicted = sum(row["weather_prediction"] for row in subset)
+                    measured = sum(row["actual"] for row in subset)
+                    factors[name] = float(np.clip((measured + 8000) / (predicted + 8000), 0.5, 1.5))
+            value.update(weights=weights, regimes=factors, member_mae_w=errors, fitted=True)
+        result[bucket] = value
+    return result
+
+
+def select_responses(
+    pairs: list[dict[str, Any]],
+    minimum_days: int,
+    minimum_samples: int,
+    improvement: float,
+    enabled_methods: tuple[str, ...] = METHODS,
+) -> dict[str, Any]:
+    """Activate only better frozen candidates, with day weighting and energy guard."""
+    result = {}
+    for bucket in LEADS:
+        part = pd.DataFrame(row for row in pairs if row["bucket"] == bucket)
+        value: dict[str, Any] = dict(
+            method="calibrated",
+            days=0,
+            samples=len(part),
+            validated=False,
+            mae_w={},
+            daily_energy_error_kwh={},
+        )
+        if not part.empty:
+            days = int(part.day.nunique())
+            errors = pd.DataFrame({method: part[method] - part.actual for method in METHODS})
+            daily_mae = errors.abs().groupby(part.day).mean()
+            mae = daily_mae.mean()
+            daily_energy = (errors.groupby(part.day).sum() * 0.25 / 1000).abs().mean()
+            enough = days >= minimum_days and len(part) >= minimum_samples
+            best = "calibrated"
+            for method in METHODS[1:]:
+                if (
+                    method in enabled_methods
+                    and enough
+                    and mae[method] < mae[best] * (1 - improvement)
+                    and daily_energy[method] <= daily_energy[best] * 1.02
+                    and (daily_mae[method] < daily_mae[best]).mean() >= 0.6
+                ):
+                    best = method
+            value.update(
+                method=best,
+                days=days,
+                validated=enough,
+                mae_w=mae.to_dict(),
+                daily_energy_error_kwh=daily_energy.to_dict(),
+            )
+        result[bucket] = value
+    return result

src/akkudoktoreos/prediction/pvakkuadaptstore.py

@@ -0,0 +1,675 @@
+"""Bounded, persistent issued PV forecasts and independently timed measurements."""
+
+import asyncio
+import hashlib
+import json
+import os
+import sqlite3
+import threading
+from collections.abc import Coroutine
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any, ClassVar, Iterator, TypeVar, cast
+
+import numpy as np
+import pandas as pd
+
+from akkudoktoreos.core.coreabc import get_config, get_database
+from akkudoktoreos.measurement.quality import MeasurementSample
+
+T = TypeVar("T")
+
+
+class _DatabaseWorker:
+    """Use the existing asynchronous EOS database from numerical worker threads."""
+
+    _loop: ClassVar[asyncio.AbstractEventLoop | None] = None
+    _lock: ClassVar[threading.Lock] = threading.Lock()
+    _pid: ClassVar[int] = 0
+
+    @classmethod
+    def run(cls, operation: Coroutine[Any, Any, T]) -> T:
+        """Run database frontend operations on a shared, process-local I/O loop."""
+        with cls._lock:
+            if cls._loop is None or cls._pid != os.getpid():
+                cls._loop = asyncio.new_event_loop()
+                cls._pid = os.getpid()
+                threading.Thread(
+                    target=cls._loop.run_forever, daemon=True, name="pv-database"
+                ).start()
+            loop = cls._loop
+        return asyncio.run_coroutine_threadsafe(operation, loop).result(timeout=120)
+
+
+class PVLearningStore:
+    """Persist immutable issue/target records through the configured EOS database.
+
+    SQLite is only an in-memory query index. Durable records live in the existing
+    EOS backend namespaces. NoDB intentionally learns in memory without persistence.
+    The old private SQLite file is read-only migration input and remains untouched.
+    """
+
+    _keys = {
+        "samples": ("key", "time_ms"),
+        "forecasts": ("signature", "issue_ms", "target_ms"),
+        "baselines": ("signature", "target_ms"),
+        "latest": ("target_ms",),
+        "metadata": ("key",),
+        "weather": ("issue_ms",),
+        "longforecasts": ("signature", "issue_ms", "target_ms"),
+    }
+
+    def __init__(self, path: Path):
+        self.path = path
+        self.database = get_database()
+        self.persistent = get_config().database.provider not in (None, "NoDB")
+        self._lock = threading.RLock()
+        self._db = sqlite3.connect(":memory:", check_same_thread=False)
+        self._db.row_factory = sqlite3.Row
+        self._initializing = True
+        self._saved_sequence = 0
+        migrated = None
+        with self.connection() as db:
+            db.executescript("""
+                CREATE TABLE IF NOT EXISTS samples (
+                    key TEXT NOT NULL, time_ms INTEGER NOT NULL, value REAL,
+                    quality TEXT NOT NULL, available_ms INTEGER NOT NULL,
+                    PRIMARY KEY(key,time_ms));
+                CREATE TABLE IF NOT EXISTS forecasts (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    signature TEXT NOT NULL, issue_ms INTEGER NOT NULL, target_ms INTEGER NOT NULL,
+                    lead_minutes REAL NOT NULL, payload TEXT NOT NULL,
+                    actual_w REAL, reference_actual_w REAL, available_ms INTEGER,
+                    updated_ms INTEGER NOT NULL,
+                    UNIQUE(signature,issue_ms,target_ms));
+                CREATE INDEX IF NOT EXISTS forecast_target ON forecasts(signature,target_ms);
+                CREATE TABLE IF NOT EXISTS baselines (
+                    signature TEXT NOT NULL, target_ms INTEGER NOT NULL,
+                    issue_ms INTEGER NOT NULL, payload TEXT NOT NULL,
+                    PRIMARY KEY(signature,target_ms));
+                CREATE TABLE IF NOT EXISTS latest (
+                    target_ms INTEGER PRIMARY KEY, issue_ms INTEGER NOT NULL, payload TEXT NOT NULL);
+                CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, payload TEXT NOT NULL);
+                CREATE TABLE IF NOT EXISTS weather (issue_ms INTEGER PRIMARY KEY, payload TEXT NOT NULL);
+                CREATE TABLE IF NOT EXISTS longforecasts (
+                    signature TEXT NOT NULL, issue_ms INTEGER NOT NULL, target_ms INTEGER NOT NULL,
+                    payload TEXT NOT NULL, PRIMARY KEY(signature,issue_ms,target_ms));
+            """)
+        if self.persistent:
+            _DatabaseWorker.run(self.database.open(namespace="pvakkuadapt"))
+            migrated = _DatabaseWorker.run(self.database.get_metadata(namespace="pvakkuadapt"))
+            if path.is_file() and not migrated:
+                self._import_legacy()
+            else:
+                self._load_backend()
+        self._track_changes()
+        self._initializing = False
+        if self.persistent and not migrated:
+            if path.is_file():
+                for table, keys in self._keys.items():
+                    expression = ",".join(keys)
+                    self._db.execute(
+                        f"INSERT OR REPLACE INTO _changes SELECT ?,json_array({expression}) FROM {table}",  # noqa: S608
+                        (table,),
+                    )  # noqa: S608
+                self._db.commit()
+            self._flush()
+            _DatabaseWorker.run(self.database.flush())
+            _DatabaseWorker.run(
+                self.database.set_metadata(
+                    self.dump(
+                        dict(
+                            schema_version=1,
+                            migrated_from=str(path) if path.is_file() else None,
+                            counts=self.counts(),
+                        )
+                    ).encode(),
+                    namespace="pvakkuadapt",
+                )
+            )
+            _DatabaseWorker.run(self.database.flush())
+
+    def _import_legacy(self) -> None:
+        """Read the old archive without changing data, IDs or availability timestamps."""
+        with sqlite3.connect(self.path.resolve().as_uri() + "?mode=ro", uri=True) as old:
+            old.row_factory = sqlite3.Row
+            for table in ("samples", "forecasts", "baselines", "latest", "metadata"):
+                cursor = old.execute("SELECT * FROM " + table)  # noqa: S608 -- fixed table list
+                while batch := cursor.fetchmany(2000):
+                    self._insert_rows(table, [dict(row) for row in batch])
+            sequence = old.execute(
+                "SELECT seq FROM sqlite_sequence WHERE name='forecasts'"
+            ).fetchone()
+            if sequence:
+                self._restore_sequence(int(sequence[0]))
+
+    def _restore_sequence(self, sequence: int) -> None:
+        current = self._db.execute(
+            "SELECT seq FROM sqlite_sequence WHERE name='forecasts'"
+        ).fetchone()
+        if current:
+            self._db.execute(
+                "UPDATE sqlite_sequence SET seq=? WHERE name='forecasts'",
+                (max(sequence, current[0]),),
+            )
+        else:
+            self._db.execute("INSERT INTO sqlite_sequence VALUES ('forecasts',?)", (sequence,))
+        self._db.commit()
+
+    def _insert_rows(self, table: str, rows: list[dict[str, Any]]) -> None:
+        if not rows:
+            return
+        if table not in self._keys:
+            raise ValueError("Unknown PV record type")
+        columns = tuple(rows[0])
+        allowed = {row[1] for row in self._db.execute("PRAGMA table_info(" + table + ")")}
+        if set(columns) != allowed:
+            raise ValueError("Invalid PV database record columns")
+        placeholders = ",".join("?" for _ in columns)
+        self._db.executemany(
+            f"INSERT OR REPLACE INTO {table} ({','.join(columns)}) VALUES ({placeholders})",  # noqa: S608
+            [tuple(row[column] for column in columns) for row in rows],
+        )
+        self._db.commit()
+
+    async def _backend_rows(self, table: str) -> list[dict[str, Any]]:
+        return [
+            json.loads(self.database.deserialize_data(value))
+            async for _, value in self.database.iterate_records(namespace="pvakkuadapt." + table)
+        ]
+
+    def _load_backend(self) -> None:
+        for table in self._keys:
+            self._insert_rows(table, _DatabaseWorker.run(self._backend_rows(table)))
+        sequence = _DatabaseWorker.run(
+            self.database.get_metadata(namespace="pvakkuadapt.forecasts")
+        )
+        if sequence:
+            self._restore_sequence(int(sequence))
+            self._saved_sequence = int(sequence)
+
+    def _track_changes(self) -> None:
+        self._db.execute(
+            "CREATE TABLE _changes (table_name TEXT, record_key TEXT, PRIMARY KEY(table_name,record_key))"
+        )
+        for table, keys in self._keys.items():
+            for event, source in (("INSERT", "NEW"), ("UPDATE", "NEW"), ("DELETE", "OLD")):
+                expression = ",".join(source + "." + key for key in keys)
+                self._db.execute(
+                    f"CREATE TRIGGER track_{table}_{event} AFTER {event} ON {table} BEGIN "  # noqa: S608
+                    f"INSERT OR REPLACE INTO _changes VALUES ('{table}',json_array({expression})); END"
+                )
+        self._db.commit()
+
+    def _flush(self) -> None:
+        if self._initializing:
+            return
+        if self.persistent:
+            for table, keys in self._keys.items():
+                changed = self._db.execute(
+                    "SELECT record_key FROM _changes WHERE table_name=?", (table,)
+                ).fetchall()
+                writes, deletes = [], []
+                for (encoded_key,) in changed:
+                    identity = json.loads(encoded_key)
+                    row = self._db.execute(
+                        f"SELECT * FROM {table} WHERE " + " AND ".join(k + "=?" for k in keys),  # noqa: S608
+                        identity,
+                    ).fetchone()  # noqa: S608
+                    key = encoded_key.encode()
+                    if row is None:
+                        deletes.append(key)
+                    else:
+                        writes.append(
+                            (key, self.database.serialize_data(self.dump(dict(row)).encode()))
+                        )
+                if writes:
+                    count = _DatabaseWorker.run(
+                        self.database.save_records(writes, namespace="pvakkuadapt." + table)
+                    )
+                    if count != len(writes):
+                        raise RuntimeError("Incomplete PV database write")
+                if deletes:
+                    _DatabaseWorker.run(
+                        self.database.delete_records(deletes, namespace="pvakkuadapt." + table)
+                    )
+            sequence = self._db.execute(
+                "SELECT seq FROM sqlite_sequence WHERE name='forecasts'"
+            ).fetchone()
+            if sequence and sequence[0] > self._saved_sequence:
+                _DatabaseWorker.run(
+                    self.database.set_metadata(
+                        str(sequence[0]).encode(), namespace="pvakkuadapt.forecasts"
+                    )
+                )
+                self._saved_sequence = int(sequence[0])
+        self._db.execute("DELETE FROM _changes")
+        self._db.commit()
+
+    def counts(self) -> dict[str, int]:
+        """Domain record counts for migration verification and diagnostics."""
+        return {
+            table: self._db.execute("SELECT COUNT(*) FROM " + table).fetchone()[0]  # noqa: S608
+            for table in self._keys
+        }  # noqa: S608 -- fixed internal table names
+
+    def storage_status(self) -> dict[str, Any]:
+        """Describe persistence explicitly rather than silently creating a private DB."""
+        return dict(
+            backend=get_config().database.provider or "NoDB",
+            persistent=self.persistent,
+            namespace="pvakkuadapt",
+            records=self.counts(),
+        )
+
+    @contextmanager
+    def connection(self) -> Iterator[sqlite3.Connection]:
+        """Use short atomic transactions and bound lock waits across ingestion/readout."""
+        with self._lock:
+            if not self._initializing:
+                self._flush()
+            with self._db:
+                yield self._db
+            self._flush()
+
+    @staticmethod
+    def milliseconds(timestamp: pd.Timestamp) -> int:
+        """Milliseconds from an explicitly zoned timestamp."""
+        value = pd.Timestamp(timestamp)
+        if value.tzinfo is None:
+            raise ValueError("Explicit timezone required.")
+        return int(value.timestamp() * 1000)
+
+    @staticmethod
+    def dump(payload: Any) -> str:
+        """Serialize finite values; missing model inputs remain null."""
+
+        def clean(value: Any) -> Any:
+            if isinstance(value, dict):
+                return {key: clean(item) for key, item in value.items()}
+            if isinstance(value, (list, tuple)):
+                return [clean(item) for item in value]
+            if isinstance(value, pd.Timestamp):
+                return value.isoformat() if pd.notna(value) else None
+            if isinstance(value, (float, np.floating)):
+                return float(value) if np.isfinite(value) else None
+            if isinstance(value, np.integer):
+                return int(value)
+            if value is pd.NaT:
+                return None
+            return value
+
+        return json.dumps(clean(payload), allow_nan=False, separators=(",", ":"))
+
+    def metadata(self, key: str, value: Any = None) -> Any:
+        """Read or atomically replace separately named model/status metadata."""
+        with self.connection() as db:
+            if value is not None:
+                db.execute(
+                    "INSERT INTO metadata VALUES (?,?) ON CONFLICT(key) DO UPDATE SET payload=excluded.payload",
+                    (key, self.dump(value)),
+                )
+            row = db.execute("SELECT payload FROM metadata WHERE key=?", (key,)).fetchone()
+        return json.loads(row[0]) if row else None
+
+    def import_samples(self, samples: list[MeasurementSample], now: pd.Timestamp) -> int:
+        """Retain the first real arrival; a changed value gets its new availability."""
+        received = self.milliseconds(now)
+        values = [
+            (
+                sample.key,
+                self.milliseconds(pd.Timestamp(sample.date_time)),
+                sample.value,
+                sample.quality.status,
+                received,
+            )
+            for sample in samples
+        ]
+        if any(time > received for _, time, *_ in values):
+            raise ValueError("Future PV measurements cannot be imported.")
+        with self.connection() as db:
+            db.executemany(
+                """INSERT INTO samples VALUES (?,?,?,?,?)
+                ON CONFLICT(key,time_ms) DO UPDATE SET value=excluded.value, quality=excluded.quality,
+                available_ms=CASE WHEN samples.value IS excluded.value AND samples.quality=excluded.quality
+                THEN samples.available_ms ELSE excluded.available_ms END""",
+                values,
+            )
+        return len(values)
+
+    def samples(self, key: str, start: pd.Timestamp, cutoff: pd.Timestamp) -> list[dict[str, Any]]:
+        """Exclude future source timestamps and samples received after the cutoff."""
+        with self.connection() as db:
+            return [
+                dict(row)
+                for row in db.execute(
+                    "SELECT * FROM samples WHERE key=? AND time_ms>=? AND time_ms<=? AND available_ms<=? ORDER BY time_ms",
+                    (
+                        key,
+                        self.milliseconds(start),
+                        self.milliseconds(cutoff),
+                        self.milliseconds(cutoff),
+                    ),
+                )
+            ]
+
+    def remember(
+        self,
+        signature: str,
+        issue: pd.Timestamp,
+        rows: list[tuple[pd.Timestamp, dict[str, Any]]],
+        timezone: str,
+    ) -> None:
+        """Archive each short lead once and freeze previous-day physical baselines."""
+        issued = self.milliseconds(issue)
+        with self.connection() as db:
+            for timestamp, payload in rows:
+                target = self.milliseconds(timestamp)
+                if target < issued:
+                    continue
+                text = self.dump({**payload, "signature": signature})
+                db.execute(
+                    "INSERT INTO latest VALUES (?,?,?) ON CONFLICT(target_ms) DO UPDATE SET issue_ms=excluded.issue_ms,payload=excluded.payload",
+                    (target, issued, text),
+                )
+                day_start = pd.Timestamp(timestamp).tz_convert(timezone).normalize()
+                existing = db.execute(
+                    "SELECT issue_ms FROM baselines WHERE signature=? AND target_ms=?",
+                    (signature, target),
+                ).fetchone()
+                if existing is None or issued < self.milliseconds(day_start):
+                    db.execute(
+                        "INSERT INTO baselines VALUES (?,?,?,?) ON CONFLICT(signature,target_ms) DO UPDATE SET issue_ms=excluded.issue_ms,payload=excluded.payload",
+                        (signature, target, issued, text),
+                    )
+                lead = (target - issued) / 60000
+                if lead < 120:
+                    db.execute(
+                        "INSERT OR IGNORE INTO forecasts(signature,issue_ms,target_ms,lead_minutes,payload,updated_ms) VALUES (?,?,?,?,?,?)",
+                        (signature, issued, target, lead, text, issued),
+                    )
+
+    def archive_weather(self, data: dict[str, Any], received: pd.Timestamp) -> str:
+        """Archive each received response once, retaining only genuinely future inputs.
+
+        Open-Meteo does not expose the numerical model's issuance here. The first
+        local receipt is recorded explicitly; returned past data never become
+        historical forecasts. Cache hits preserve that first receipt.
+        """
+        digest = hashlib.sha256(self.dump(data).encode()).hexdigest()
+        with self.connection() as db:
+            if db.execute(
+                "SELECT 1 FROM weather WHERE json_extract(payload,'$.digest')=? LIMIT 1", (digest,)
+            ).fetchone():
+                return digest
+            raw = data["minutely_15"]
+            times = pd.DatetimeIndex(pd.to_datetime(raw["time"], utc=True))
+            starts = times - pd.Timedelta(minutes=15)
+            keep = (starts >= received) & (starts <= received + pd.Timedelta(hours=48))
+            subset = {
+                key: [value for value, valid in zip(values, keep) if valid]
+                for key, values in raw.items()
+            }
+            payload = dict(
+                digest=digest,
+                received_at=received.isoformat(),
+                source_issue_at=None,
+                timestamp_semantics="interval_end",
+                minutely_15=subset,
+                units=data.get("minutely_15_units", {}),
+                elevation=data.get("elevation"),
+            )
+            db.execute(
+                "INSERT OR IGNORE INTO weather VALUES (?,?)",
+                (self.milliseconds(received), self.dump(payload)),
+            )
+        return digest
+
+    def remember_long(
+        self, signature: str, issue: pd.Timestamp, rows: list[tuple[pd.Timestamp, dict[str, Any]]]
+    ) -> None:
+        """Freeze an hourly snapshot per future target, bounding archive growth."""
+        issued = self.milliseconds(issue)
+        bucket = self.milliseconds(issue.floor("h"))
+        with self.connection() as db:
+            for timestamp, payload in rows:
+                target = self.milliseconds(timestamp)
+                if not issued < target <= issued + 48 * 3600000:
+                    continue
+                db.execute(
+                    "INSERT OR IGNORE INTO longforecasts VALUES (?,?,?,?)",
+                    (
+                        signature,
+                        bucket,
+                        target,
+                        self.dump({**payload, "issued_ms": issued, "response_version": 1}),
+                    ),
+                )
+
+    def long_rows(
+        self, signature: str, start: pd.Timestamp, cutoff: pd.Timestamp
+    ) -> list[dict[str, Any]]:
+        """Read issued long-horizon candidates, with no reconstructed past weather."""
+        with self.connection() as db:
+            rows = db.execute(
+                "SELECT * FROM longforecasts WHERE signature=? AND target_ms>=? AND target_ms<? ORDER BY target_ms,issue_ms",
+                (signature, self.milliseconds(start), self.milliseconds(cutoff)),
+            ).fetchall()
+        return [{**dict(row), **json.loads(row["payload"]), "payload": None} for row in rows]
+
+    def resolve(self, signature: str, observed: pd.DataFrame, now: pd.Timestamp) -> None:
+        """Attach first complete observations, preserving the original forecast."""
+        updated = self.milliseconds(now)
+        with self.connection() as db:
+            for timestamp, values in observed.iterrows():
+                target = self.milliseconds(cast(pd.Timestamp, timestamp))
+                if target + 900000 > updated:
+                    continue
+                if (
+                    pd.notna(values.actual_w)
+                    and np.isfinite(float(values.actual_w))
+                    and pd.notna(values.available_at)
+                ):
+                    if self.milliseconds(values.available_at) <= updated:
+                        db.execute(
+                            "UPDATE forecasts SET actual_w=?,updated_ms=? WHERE signature=? AND target_ms=? AND actual_w IS NULL",
+                            (float(values.actual_w), updated, signature, target),
+                        )
+                if pd.isna(values.reference_actual_w) or not np.isfinite(
+                    float(values.reference_actual_w)
+                ):
+                    continue
+                availability = self.milliseconds(values.available_at)
+                if availability > updated or target + 900000 > updated:
+                    continue
+                actual = (
+                    float(values.actual_w)
+                    if pd.notna(values.actual_w) and np.isfinite(float(values.actual_w))
+                    else None
+                )
+                db.execute(
+                    """UPDATE forecasts SET reference_actual_w=?,actual_w=?,available_ms=?,updated_ms=?
+                    WHERE signature=? AND target_ms=? AND reference_actual_w IS NULL""",
+                    (
+                        float(values.reference_actual_w),
+                        actual,
+                        availability,
+                        updated,
+                        signature,
+                        target,
+                    ),
+                )
+                if actual is not None:
+                    db.execute(
+                        "UPDATE forecasts SET actual_w=?,updated_ms=? WHERE signature=? AND target_ms=? AND actual_w IS NULL AND reference_actual_w IS NOT NULL",
+                        (actual, updated, signature, target),
+                    )
+
+    def training(self, signature: str, cutoff: pd.Timestamp, days: int = 21) -> pd.DataFrame:
+        """Return archived inputs with independently available reference targets."""
+        end = self.milliseconds(cutoff)
+        with self.connection() as db:
+            rows = db.execute(
+                "SELECT * FROM forecasts WHERE signature=? AND issue_ms>=? AND available_ms<=? ORDER BY issue_ms,target_ms",
+                (signature, end - days * 86400000, end),
+            ).fetchall()
+        result = []
+        for row in rows:
+            values = json.loads(row["payload"])
+            if values.get("features") is None or not values.get("reference_sky_w"):
+                continue
+            result.append(
+                {
+                    **values["features"],
+                    "issue": pd.Timestamp(row["issue_ms"], unit="ms", tz="UTC"),
+                    "time": pd.Timestamp(row["target_ms"], unit="ms", tz="UTC"),
+                    "lead_minutes": row["lead_minutes"],
+                    "target_available_at": pd.Timestamp(row["available_ms"], unit="ms", tz="UTC"),
+                    "target_residual": (
+                        row["reference_actual_w"] - values["reference_calibrated_w"]
+                    )
+                    / values["reference_sky_w"],
+                    "reference_sky_w": values["reference_sky_w"],
+                }
+            )
+        return pd.DataFrame(result)
+
+    def baselines(
+        self, signature: str, start: pd.Timestamp, cutoff: pd.Timestamp
+    ) -> list[tuple[pd.Timestamp, dict[str, Any]]]:
+        """Return genuine issued physical baselines only, never revised weather."""
+        with self.connection() as db:
+            rows = db.execute(
+                "SELECT * FROM baselines WHERE signature=? AND target_ms>=? AND target_ms<? AND issue_ms<=target_ms ORDER BY target_ms",
+                (signature, self.milliseconds(start), self.milliseconds(cutoff)),
+            ).fetchall()
+        return [
+            (pd.Timestamp(row["target_ms"], unit="ms", tz="UTC"), json.loads(row["payload"]))
+            for row in rows
+        ]
+
+    def comparisons(
+        self, signature: str, start_ms: int, end_ms: int, limit: int = 2000, since_ms: int = 0
+    ) -> list[dict[str, Any]]:
+        """Export bounded matched forecasts, keeping issue time and lead for Grafana."""
+        with self.connection() as db:
+            rows = db.execute(
+                "SELECT * FROM forecasts WHERE signature=? AND target_ms>=? AND target_ms<? AND updated_ms>=? AND (reference_actual_w IS NOT NULL OR actual_w IS NOT NULL) ORDER BY updated_ms,id LIMIT ?",
+                (signature, start_ms, end_ms, since_ms, limit),
+            ).fetchall()
+        return [{**dict(row), **json.loads(row["payload"]), "payload": None} for row in rows]
+
+    def validation_rows(
+        self, signature: str, start: pd.Timestamp, end: pd.Timestamp
+    ) -> list[dict[str, Any]]:
+        """Read the complete validation window; public export pagination is separate."""
+        with self.connection() as db:
+            rows = db.execute(
+                "SELECT issue_ms,target_ms,lead_minutes,payload FROM forecasts WHERE signature=? AND target_ms>=? AND target_ms<? ORDER BY target_ms,issue_ms",
+                (signature, self.milliseconds(start), self.milliseconds(end)),
+            ).fetchall()
+        return [{**dict(row), **json.loads(row["payload"]), "payload": None} for row in rows]
+
+    def metric_rows(
+        self, signature: str, start: pd.Timestamp, end: pd.Timestamp
+    ) -> list[dict[str, Any]]:
+        """Full-window scalar projection avoids decoding every feature/model payload."""
+        with self.connection() as db:
+            rows = db.execute(
+                """SELECT target_ms,lead_minutes,actual_w,reference_actual_w,
+                COALESCE(json_extract(payload,'$.algorithm_version'),1) AS algorithm_version,
+                json_extract(payload,'$.plant_raw_w') AS plant_raw_w,
+                json_extract(payload,'$.plant_calibrated_w') AS plant_calibrated_w,
+                json_extract(payload,'$.plant_local_w') AS plant_local_w,
+                json_extract(payload,'$.plant_adaptive_w') AS plant_adaptive_w,
+                json_extract(payload,'$.reference_raw_w') AS reference_raw_w,
+                json_extract(payload,'$.reference_calibrated_w') AS reference_calibrated_w,
+                json_extract(payload,'$.reference_local_w') AS reference_local_w,
+                json_extract(payload,'$.reference_adaptive_w') AS reference_adaptive_w
+                FROM forecasts WHERE signature=? AND target_ms>=? AND target_ms<?
+                AND (actual_w IS NOT NULL OR reference_actual_w IS NOT NULL)""",
+                (signature, self.milliseconds(start), self.milliseconds(end)),
+            ).fetchall()
+        return [dict(row) for row in rows]
+
+    def daily_metrics(
+        self, signature: str, start: pd.Timestamp, end: pd.Timestamp, timezone: str
+    ) -> list[dict[str, Any]]:
+        """Evaluate frozen pre-midnight forecasts only against complete local days."""
+        begin, until = self.milliseconds(start), self.milliseconds(end)
+        with self.connection() as db:
+            rows = db.execute(
+                """SELECT b.target_ms,b.issue_ms,b.payload,o.actual_w FROM baselines b
+                LEFT JOIN (SELECT target_ms,MAX(actual_w) actual_w FROM forecasts
+                WHERE signature=? AND target_ms>=? AND target_ms<? GROUP BY target_ms) o
+                ON b.target_ms=o.target_ms WHERE b.signature=? AND b.target_ms>=? AND b.target_ms<? ORDER BY b.target_ms""",
+                (signature, begin, until, signature, begin, until),
+            ).fetchall()
+        days: dict[str, list[dict[str, Any]]] = {}
+        for row in rows:
+            time = pd.Timestamp(row["target_ms"], unit="ms", tz="UTC").tz_convert(timezone)
+            days.setdefault(str(time.date()), []).append(
+                {**dict(row), **json.loads(row["payload"])}
+            )
+        result = []
+        for day, part in days.items():
+            left = pd.Timestamp(day, tz=timezone)
+            right = left + pd.DateOffset(days=1)
+            expected = int((right - left).total_seconds() / 900)
+            complete = (
+                right <= end
+                and len(part) == expected
+                and all(
+                    r["actual_w"] is not None and r["issue_ms"] < self.milliseconds(left)
+                    for r in part
+                )
+            )
+            result.append(
+                dict(
+                    day=day,
+                    complete=bool(complete),
+                    slots=len(part),
+                    expected_slots=expected,
+                    observed_slots=sum(r["actual_w"] is not None for r in part),
+                    actual_kwh=sum(r["actual_w"] for r in part) * 0.25 / 1000 if complete else None,
+                    **{
+                        method + "_kwh": sum(r["plant_" + method + "_w"] for r in part)
+                        * 0.25
+                        / 1000
+                        if complete
+                        and all(r.get("plant_" + method + "_w") is not None for r in part)
+                        else None
+                        for method in ("raw", "calibrated", "local", "adaptive")
+                    },
+                )
+            )
+        return result
+
+    def latest(self, start_ms: int) -> list[dict[str, Any]]:
+        """Current public curve, distinct from the immutable validation archive."""
+        with self.connection() as db:
+            rows = db.execute(
+                "SELECT * FROM latest WHERE target_ms>=? ORDER BY target_ms LIMIT 1600", (start_ms,)
+            ).fetchall()
+        return [
+            {
+                "target_ms": row["target_ms"],
+                "issue_ms": row["issue_ms"],
+                **json.loads(row["payload"]),
+            }
+            for row in rows
+        ]
+
+    def prune(self, cutoff: pd.Timestamp, days: int = 60) -> None:
+        """Bound only this provider's private archive; ordinary EOS data is untouched."""
+        oldest = self.milliseconds(cutoff) - days * 86400000
+        with self.connection() as db:
+            for statement in (
+                "DELETE FROM samples WHERE time_ms<?",
+                "DELETE FROM forecasts WHERE target_ms<?",
+                "DELETE FROM baselines WHERE target_ms<?",
+                "DELETE FROM latest WHERE target_ms<?",
+                "DELETE FROM weather WHERE issue_ms<?",
+                "DELETE FROM longforecasts WHERE target_ms<?",
+            ):
+                db.execute(statement, (oldest,))

src/akkudoktoreos/prediction/pvakkuadapttrend.py

@@ -0,0 +1,248 @@
+"""Causal, lead-specific learning of local PV residual persistence and trends.
+
+Training needs archived features from genuinely issued forecasts, plus the
+later availability of their independent reference measurements. A revised
+weather history cannot substitute for that archive.
+"""
+
+from dataclasses import dataclass, field
+from typing import cast
+
+import numpy as np
+import pandas as pd
+
+from akkudoktoreos.prediction.pvakkuadaptmodel import _aware, _aware_index
+
+LEVEL_FEATURES = ("prior_residual", "level", "weather_change", "intercept")
+TREND_FEATURES = ("ramp", "slope", "volatility", "consistent_ramp")
+FEATURES = LEVEL_FEATURES + TREND_FEATURES
+
+
+def local_trend_features(
+    history: pd.DataFrame,
+    future_reference: pd.Series,
+    future_reference_sky: pd.Series,
+    cutoff: pd.Timestamp,
+    reference_interval_minutes: int = 15,
+) -> pd.DataFrame:
+    """Describe four complete past intervals without reading future actuals.
+
+    Slopes describe deviations from the current weather model, normalized by the
+    reference's clear-sky power. They therefore exclude its regular daily solar
+    curve. All future quantities must come from a weather run available at cutoff.
+    An invalid latest interval disables features instead of selecting older values.
+    """
+    cutoff = _aware(cutoff)
+    future_index = _aware_index(future_reference.index)
+    if reference_interval_minutes not in (5, 15):
+        raise ValueError("Reference resolution must be 5 or 15 minutes.")
+    _aware_index(history.index, reference_interval_minutes)
+    if history.index.has_duplicates or future_index.has_duplicates:
+        raise ValueError("Distinct reference and forecast timestamps required.")
+    result = pd.DataFrame(np.nan, index=future_index, columns=list(FEATURES))
+    result["feature_available_at"] = pd.Series(
+        pd.NaT, index=future_index, dtype="datetime64[ns, UTC]"
+    )
+    if history.empty:
+        return result
+    available = pd.to_datetime(history["available_at"], utc=True)
+    recent = (
+        history[
+            (available <= cutoff)
+            & (history.index + pd.Timedelta(minutes=reference_interval_minutes) <= cutoff)
+            & (history.index >= cutoff - pd.Timedelta(hours=2))
+        ]
+        .sort_index()
+        .tail(4)
+    )
+    if len(recent) != 4:
+        return result
+    reference_index = _aware_index(recent.index, reference_interval_minutes)
+    interval = pd.Timedelta(minutes=reference_interval_minutes)
+    if not ((reference_index[1:] - reference_index[:-1]) == interval).all():
+        return result
+    if cutoff - (recent.index[-1] + interval) > pd.Timedelta(minutes=30):
+        return result
+    values = recent[["calibrated_power", "measured_power", "clear_sky_power"]]
+    if not np.isfinite(values.to_numpy()).all():
+        return result
+    if (values[["calibrated_power", "measured_power"]] < 0).any().any():
+        return result
+    if (values.clear_sky_power < 150).any():
+        return result
+    if "eligible" in recent and not recent.eligible.fillna(False).all():
+        return result
+    residual = (
+        ((values.measured_power - values.calibrated_power) / values.clear_sky_power)
+        .clip(-1.5, 1.5)
+        .to_numpy(dtype=float)
+    )
+    level = float(np.dot(residual[-2:], [0.35, 0.65]))
+    changes = np.diff(residual) * 15 / reference_interval_minutes
+    ramp = float(changes[-1])
+    slope = float(np.dot(residual, [-0.3, -0.1, 0.1, 0.3]) * 15 / reference_interval_minutes)
+    volatility = float(np.std(changes))
+    consistent = ramp if np.sign(changes[-1]) == np.sign(changes[-2]) else 0.0
+    current_weather = float(values.calibrated_power.iloc[-1] / values.clear_sky_power.iloc[-1])
+    sky = future_reference_sky.reindex(future_index)
+    lead = (future_index - cutoff).total_seconds().to_numpy() / 60
+    age = (cutoff - (recent.index[-1] + interval / 2)).total_seconds() / 60
+    decay = np.exp2(-np.maximum(0, lead + age) / 30)
+    taper = np.clip(2 * (1 - lead / 120), 0, 1)
+    valid = (
+        (lead >= 0)
+        & (lead < 120)
+        & (future_index.tz_convert(cutoff.tz).date == cutoff.date())
+        & sky.ge(150).to_numpy()
+        & np.isfinite(future_reference.to_numpy())
+        & future_reference.ge(0).to_numpy()
+    )
+    rows = pd.DataFrame(
+        {
+            "prior_residual": level * decay * taper,
+            "level": level,
+            "weather_change": (current_weather - future_reference / sky).clip(-1.5, 1.5),
+            "intercept": 1.0,
+            "ramp": ramp,
+            "slope": slope,
+            "volatility": volatility,
+            "consistent_ramp": consistent,
+            "feature_available_at": pd.to_datetime(recent.available_at, utc=True).max(),
+        },
+        index=future_index,
+    )
+    result.loc[valid, :] = rows.loc[valid, :]
+    return result
+
+
+@dataclass
+class LocalPVTrendModel:
+    """Learn a robust correction per forecast lead, shrinking to the old nowcast.
+
+    Each daily fit uses only issued samples whose target measurements were already
+    available on completed earlier local days. Refit coefficients from that archive
+    instead of training on corrected forecasts or curtailed whole-plant production.
+    """
+
+    use_trend: bool = True
+    timezone: str = "Europe/Berlin"
+    learning_days: int = 21
+    half_life_days: float = 14.0
+    min_days: int = 5
+    min_samples: int = 100
+    ridge: float = 20.0
+    scale_features: bool = False
+    coefficients: dict[int, np.ndarray] = field(default_factory=dict, init=False)
+    training_counts: dict[int, int] = field(default_factory=dict, init=False)
+
+    @property
+    def feature_names(self) -> tuple[str, ...]:
+        """Features shared with the level-only comparison and optional trends."""
+        return FEATURES if self.use_trend else LEVEL_FEATURES
+
+    def fit(self, samples: pd.DataFrame, cutoff: pd.Timestamp) -> "LocalPVTrendModel":
+        """Fit using available labels from previous days, with recency and Huber weights."""
+        cutoff = _aware(cutoff)
+        self.coefficients, self.training_counts = {}, {}
+        if self.ridge <= 0 or self.half_life_days <= 0:
+            raise ValueError("Positive regularization and recency half-life required.")
+        if samples.empty:
+            return self
+        issued = pd.to_datetime(samples.issue, utc=True)
+        target_time = pd.to_datetime(samples.time, utc=True)
+        available = pd.to_datetime(samples.target_available_at, utc=True)
+        feature_available = pd.to_datetime(samples.feature_available_at, utc=True)
+        today = cutoff.tz_convert(self.timezone).normalize()
+        target_day = target_time.dt.tz_convert(self.timezone).dt.normalize()
+        valid = (
+            (issued < target_time)
+            & (feature_available <= issued)
+            & (available >= target_time + pd.Timedelta(minutes=15))
+            & (available <= cutoff)
+            & (target_day < today)
+            & (issued >= cutoff - pd.Timedelta(days=self.learning_days))
+            & pd.Series(
+                np.isfinite(
+                    samples[[*self.feature_names, "target_residual", "reference_sky_w"]].to_numpy(
+                        dtype=float
+                    )
+                ).all(axis=1),
+                index=samples.index,
+            )
+            & samples.reference_sky_w.ge(150)
+            & samples.lead_minutes.between(0, 120, inclusive="left")
+        )
+        selected = samples.loc[valid].copy()
+        if selected.empty:
+            return self
+        selected["bucket"] = np.ceil(selected.lead_minutes / 15).astype(int) * 15
+        prior = np.zeros(len(self.feature_names))
+        prior[0] = 1.0
+        for bucket, part in selected.groupby("bucket"):
+            bucket_number = cast(int, bucket)
+            count = len(part)
+            days = (
+                pd.to_datetime(part.time, utc=True).dt.tz_convert(self.timezone).dt.date.nunique()
+            )
+            self.training_counts[bucket_number] = count
+            if count < self.min_samples or days < self.min_days:
+                continue
+            x = part[list(self.feature_names)].to_numpy(dtype=float)
+            y = part.target_residual.clip(-1.5, 1.5).to_numpy(dtype=float)
+            age = (cutoff - pd.to_datetime(part.issue, utc=True)).dt.total_seconds() / 86400
+            base = np.exp2(-age.to_numpy() / self.half_life_days)
+            base *= np.minimum(1, part.reference_sky_w.to_numpy(dtype=float) / 1500)
+            weight = base.copy()
+            scale = np.ones(len(prior))
+            if self.scale_features:
+                # Penalize standardized feature contributions. A physically
+                # small slope must not be suppressed merely because of its units.
+                scale = np.maximum(np.sqrt(np.average(x * x, axis=0, weights=base)), 1e-3)
+            penalty = np.diag(scale * scale) * self.ridge
+            coef = prior.copy()
+            for _ in range(4):
+                coef = np.linalg.solve(
+                    x.T @ (weight[:, None] * x) + penalty, x.T @ (weight * y) + penalty @ prior
+                )
+                error = np.abs(y - x @ coef)
+                weight = base * np.minimum(1, 0.15 / np.maximum(error, 1e-9))
+            self.coefficients[bucket_number] = coef
+        return self
+
+    def predict(
+        self,
+        prediction: pd.Series,
+        clear_sky_power: pd.Series,
+        features: pd.DataFrame,
+        cutoff: pd.Timestamp,
+        fallback: pd.Series,
+        capacity_w: float = 0.0,
+    ) -> pd.Series:
+        """Transfer learned reference residuals to the plant; otherwise keep fallback."""
+        cutoff = _aware(cutoff)
+        index = _aware_index(prediction.index)
+        result = fallback.reindex(index).copy()
+        rows = features.reindex(index)
+        sky = clear_sky_power.reindex(index)
+        leads = (index - cutoff).total_seconds().to_numpy() / 60
+        buckets = np.ceil(leads / 15).astype(int) * 15
+        available = pd.to_datetime(rows.feature_available_at, utc=True)
+        valid = (
+            (leads >= 0)
+            & (leads < 120)
+            & (index.tz_convert(cutoff.tz).date == cutoff.date())
+            & np.isfinite(rows[list(self.feature_names)].to_numpy(dtype=float)).all(axis=1)
+            & (available <= cutoff).to_numpy()
+            & np.isfinite(sky.to_numpy())
+            & np.isfinite(prediction.to_numpy())
+        )
+        for bucket, coef in self.coefficients.items():
+            mask = valid & (buckets == bucket)
+            if not mask.any():
+                continue
+            residual = rows.loc[mask, list(self.feature_names)].to_numpy(dtype=float) @ coef
+            residual = np.clip(residual, -1.5, 1.5)
+            # Smoothly remove any learned constant near the finite horizon.
+            taper = np.clip((120 - leads[mask]) / 30, 0, 1)
+            result.loc[mask] = prediction.loc[mask] + sky.loc[mask].clip(lower=0) * residual * taper
+        return result.clip(lower=0, upper=capacity_w if capacity_w > 0 else None)

src/akkudoktoreos/prediction/pvforecast.py

@@ -6,6 +6,9 @@ from pydantic import Field, computed_field, field_validator, model_validator
 
 from akkudoktoreos.config.configabc import SettingsBaseModel
 from akkudoktoreos.core.coreabc import get_prediction
+from akkudoktoreos.prediction.pvakkuadapt import (
+    PVForecastAdaptiveAkkudoktorCommonSettings,
+)
 from akkudoktoreos.prediction.pvforecastabc import PVForecastProvider
 from akkudoktoreos.prediction.pvforecastakkudoktorlocal import (
     PVForecastAkkudoktorLocalCommonSettings,
@@ -32,6 +35,7 @@ def pvforecast_provider_ids() -> list[str]:
         # Prediction may not be initialized
         # Return at least provider used in example
         return [
+            "PVForecastAdaptiveAkkudoktor",
             "PVForecastAkkudoktor",
             "PVForecastForecastSolar",
             "PVForecastHomeAssistant",
@@ -52,6 +56,54 @@ def pvforecast_provider_ids() -> list[str]:
 class PVForecastPlaneSetting(SettingsBaseModel):
     """PV Forecast Plane Configuration."""
 
+    name: Optional[str] = Field(
+        default=None, description="Readable PV string name.", examples=[None]
+    )
+    measurement_key: Optional[str] = Field(
+        default=None,
+        description="Matching EOS power measurement key for adaptive learning.",
+        examples=[None],
+    )
+    nowcast_reference: bool = Field(
+        default=False, description="Use this independent array as a local weather reference."
+    )
+    curtailment_sensitive: bool = Field(
+        default=False,
+        description="Battery-full curtailment possible; exclude from unrestricted weather learning.",
+    )
+    parameters_estimated: bool = Field(
+        default=False,
+        description="Geometry and peak power are effective fitted model estimates, not verified nameplate data.",
+    )
+    correction_group: Optional[str] = Field(
+        default=None,
+        examples=[None],
+        min_length=1,
+        max_length=128,
+        description="Optional group of planes sharing short-term validation. Unset independent references share one group; other planes are separate.",
+    )
+    curtailment_key: Optional[str] = Field(
+        default=None,
+        examples=[None],
+        min_length=1,
+        max_length=128,
+        description="Optional measured flag: 0 explicitly unrestricted, 1 curtailed. Missing or stale flags never authorize learning a curtailment-sensitive plane.",
+    )
+    learning_guard_key: Optional[str] = Field(
+        default=None,
+        examples=[None],
+        min_length=1,
+        max_length=128,
+        description="Optional site-specific positive learning guard, e.g. battery SoC, as an alternative to a binary curtailment flag.",
+    )
+    learning_guard_max: Optional[float] = Field(
+        default=None,
+        examples=[None],
+        allow_inf_nan=False,
+        description="Learn only while every supported guard value is below this bound. This is an explicit site assumption, not automatic curtailment detection.",
+    )
+    learning_guard_max_gap_seconds: int = Field(default=600, ge=1, le=3600)
+
     # latitude: Optional[float] = Field(default=None, json_schema_extra={ "description": "Latitude in decimal degrees, between -90 and 90, north is positive (ISO 19115) (°)" })
     surface_tilt: Optional[float] = Field(
         default=30.0,
@@ -168,6 +220,15 @@ class PVForecastPlaneSetting(SettingsBaseModel):
 
     @model_validator(mode="after")
     def validate_list_length(self) -> Self:
+        if bool(self.learning_guard_key) != (self.learning_guard_max is not None):
+            raise ValueError("A learning guard requires both a key and an explicit upper bound.")
+        if self.curtailment_key and self.learning_guard_key:
+            raise ValueError("Use either a curtailment flag or a site-specific learning guard.")
+        if self.measurement_key and self.measurement_key in (
+            self.curtailment_key,
+            self.learning_guard_key,
+        ):
+            raise ValueError("Power and learning-status measurements must use distinct keys.")
         # Check if either attribute is set and add to active planes
         if self.trackingtype == 2:
             # Tilt angle from horizontal plane is ignored for two-axis tracking.
@@ -213,6 +274,11 @@ class PVForecastCommonSettings(SettingsBaseModel):
         },
     )
 
+    adaptive: PVForecastAdaptiveAkkudoktorCommonSettings = Field(
+        default_factory=PVForecastAdaptiveAkkudoktorCommonSettings,
+        description="Adaptive Akkudoktor: configure string measurements and independent reference flags on the planes.",
+    )
+
     pvforecastimport: PVForecastImportCommonSettings = Field(
         default_factory=PVForecastImportCommonSettings,
         json_schema_extra={"description": "PV forecast import provider settings"},

src/akkudoktoreos/server/eos.py

@@ -80,6 +80,7 @@ from akkudoktoreos.server.rest.error import (
     register_problem_handlers,
 )
 from akkudoktoreos.server.rest.measurement import router as measurement_router
+from akkudoktoreos.server.rest.pvakkuadaptrest import router as pvadaptive_router
 from akkudoktoreos.server.rest.starteosdash import supervise_eosdash
 from akkudoktoreos.server.retentionmanager import RetentionManager
 from akkudoktoreos.server.server import (
@@ -262,6 +263,7 @@ The genetic optimization API fields were renamed from German to English. For bac
 # ----------------------
 
 app.include_router(measurement_router)
+app.include_router(pvadaptive_router)
 
 register_problem_handlers(app)
 

src/akkudoktoreos/server/rest/pvakkuadaptrest.py

@@ -0,0 +1,93 @@
+"""Learning status, issued comparisons and bounded raw PV ingestion for Grafana."""
+
+import asyncio
+from typing import Any
+
+import pandas as pd
+from fastapi import APIRouter, Body, HTTPException, Query
+
+from akkudoktoreos.core.coreabc import get_prediction
+from akkudoktoreos.measurement.quality import MeasurementSample
+from akkudoktoreos.prediction.pvakkuadapt import (
+    PVForecastAdaptiveAkkudoktor,
+)
+
+router = APIRouter(prefix="/v1/prediction/pv-adaptive", tags=["prediction"])
+
+
+def provider() -> PVForecastAdaptiveAkkudoktor:
+    """Resolve the registered singleton without creating a separate predictor."""
+    return next(
+        p for p in get_prediction().providers if isinstance(p, PVForecastAdaptiveAkkudoktor)
+    )
+
+
+@router.get("/status")
+async def status() -> dict[str, Any]:
+    """Status contains honest cold-start counts and matched per-lead error metrics."""
+    store = await asyncio.to_thread(provider().store)
+    value = await asyncio.to_thread(store.metadata, "status")
+    return value or dict(
+        state="initializing",
+        state_code=0,
+        training_days=0,
+        trained_leads=0,
+        message="Noch keine adaptive Prognose ausgegeben.",
+        metrics=[],
+    )
+
+
+@router.put("/samples")
+async def samples(values: list[MeasurementSample] = Body(max_length=10000)) -> dict[str, int]:
+    """Store original power readings in the bounded adaptive archive, without zero filling."""
+    instance = provider()
+    allowed = instance._sample_keys()
+    if any(sample.key not in allowed for sample in values):
+        raise HTTPException(422, "Measurement key is not assigned to an adaptive PV string.")
+    if any(sample.value is not None and sample.value < 0 for sample in values):
+        raise HTTPException(422, "PV production must not be negative.")
+    flags = {
+        plane.curtailment_key for plane in instance._learning_planes() if plane.curtailment_key
+    }
+    if any(sample.key in flags and sample.value not in (None, 0, 1) for sample in values):
+        raise HTTPException(
+            422, "Curtailment flags must be 0 (unrestricted), 1 (curtailed), or null."
+        )
+    try:
+        store = await asyncio.to_thread(instance.store)
+        updated = await asyncio.to_thread(store.import_samples, values, pd.Timestamp.now(tz="UTC"))
+        return dict(updated=updated)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+
+
+@router.get("/comparisons")
+async def comparisons(
+    days: int = Query(default=2, ge=1, le=30),
+    limit: int = Query(default=2000, ge=1, le=20000),
+    since_ms: int = Query(default=0, ge=0),
+) -> list[dict[str, Any]]:
+    """Export genuinely issued forecasts with later measured truth, issuance and lead."""
+    instance = provider()
+    store = await asyncio.to_thread(instance.store)
+    state = await asyncio.to_thread(store.metadata, "status")
+    if not state:
+        return []
+    now = pd.Timestamp.now(tz="UTC")
+    return await asyncio.to_thread(
+        store.comparisons,
+        state["signature"],
+        store.milliseconds(now - pd.Timedelta(days=days)),
+        store.milliseconds(now),
+        limit,
+        since_ms,
+    )
+
+
+@router.get("/latest")
+async def latest() -> list[dict[str, Any]]:
+    """Latest complete output curve; validation uses the separate immutable archive."""
+    instance = provider()
+    store = await asyncio.to_thread(instance.store)
+    start = pd.Timestamp.now(tz=instance.config.general.timezone).normalize()
+    return await asyncio.to_thread(store.latest, store.milliseconds(start))