Add an optional export energy budget: only export battery energy that came from PV (export_budget_mode)

.cspell/custom-dictionary-workspace.txt

@@ -514,6 +514,7 @@ predheat
 preseed
 preseeded
 prevs
+prosumer
 protobuf
 pseudonymisation
 pseudonymise

apps/predbat/annual.py

@@ -750,6 +750,7 @@ def reset_sample_state(predbat):
     predbat.soc_kw = 0.0
 
     predbat.cost_today_sofar = 0
+    predbat.export_budget_seed = None
     predbat.carbon_today_sofar = 0
     predbat.iboost_today = 0
     predbat.import_today_now = 0

apps/predbat/config.py

@@ -2604,6 +2604,8 @@
     "db_primary": {"type": "boolean"},
     "threads": {"type": "string|integer", "allowed": ["auto", 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20]},
     "prediction_kernel_enable": {"type": "boolean"},
+    "export_budget_mode": {"type": "string|boolean", "allowed": ["off", "pv_origin", "pv_equivalent"]},
+    "export_budget_cap_kwh": {"type": "float", "zero": True},
     "log_count": {"type": "integer", "min": 2, "max": 100},
     "ha_url": {"type": "string", "empty": False},
     "ha_key": {"type": "string", "empty": False},

apps/predbat/export_budget.py

@@ -0,0 +1,211 @@
+# -----------------------------------------------------------------------------
+# Predbat Home Battery System
+# Copyright Trefor Southwell 2026 - All Rights Reserved
+# This application maybe used for personal use only and not for commercial use
+# -----------------------------------------------------------------------------
+# fmt off
+# pylint: disable=consider-using-f-string
+# pylint: disable=line-too-long
+# pylint: disable=attribute-defined-outside-init
+"""Export energy budget (export_budget_mode).
+
+Limits battery-to-grid export to energy derived from the household's own PV:
+
+- pv_origin: stored energy is split into a PV part and a grid part. Battery output up to the whole
+  on-site load counts as the house's own use and draws the grid part first, even while PV is exported.
+  Only battery output beyond the load is export, and only the PV part may be exported.
+- pv_equivalent: a bucket filled by PV production and drained by all export, capped (default the
+  battery size); battery export is allowed while the bucket has energy.
+
+The step functions are the reference the prediction engines follow. prediction_kernel.cpp mirrors
+them inline, so the arithmetic must not change without a kernel parity revision bump.
+"""
+
+import math
+
+EXPORT_BUDGET_OFF = 0
+EXPORT_BUDGET_PV_ORIGIN = 1
+EXPORT_BUDGET_PV_EQUIVALENT = 2
+EXPORT_BUDGET_MODE_NAMES = {"off": EXPORT_BUDGET_OFF, "pv_origin": EXPORT_BUDGET_PV_ORIGIN, "pv_equivalent": EXPORT_BUDGET_PV_EQUIVALENT}
+EXPORT_BUDGET_REPLAY_DAYS = 7
+# A 5 minute meter delta above this (240 kW) is a counter glitch, not energy
+EXPORT_BUDGET_MAX_STEP_KWH = 20.0
+
+
+def parse_export_budget_mode(value):
+    """Return the mode constant for an apps.yaml value, or None when it is not a known mode"""
+    if value is None or value is False:
+        return EXPORT_BUDGET_OFF
+    if not isinstance(value, str):
+        return None
+    return EXPORT_BUDGET_MODE_NAMES.get(value.strip().lower())
+
+
+def export_budget_house_from_battery(mode, battery_out, load, pv):
+    """Battery output (AC) that counts as serving on-site load rather than as export.
+
+    pv_origin counts the whole load, so stored grid energy may run the house while PV is exported;
+    pv_equivalent counts PV as covering the load first, so only what PV leaves uncovered.
+    """
+    if mode == EXPORT_BUDGET_PV_ORIGIN:
+        return min(battery_out, max(load, 0.0))
+    return min(battery_out, max(load - pv, 0.0))
+
+
+def export_budget_battery_export_allowed(mode, bucket, pv_part, grid_part, pv_now, pv_direct_export, house_from_battery_ac, stored_to_ac):
+    """AC energy the battery may export to the grid in one forced export step.
+
+    pv_equivalent: total export <= bucket + pv_now, so the battery may add what is left after the PV
+    exported directly. pv_origin: on-site load is served from the grid part first and export only from
+    the PV part that remains; stored_to_ac converts stored kWh to AC at the meter.
+    """
+    if mode == EXPORT_BUDGET_PV_EQUIVALENT:
+        return max(bucket + pv_now - pv_direct_export, 0.0)
+    house_stored = house_from_battery_ac / stored_to_ac if stored_to_ac > 0 else 0.0
+    pv_left = max(pv_part - max(house_stored - grid_part, 0.0), 0.0)
+    return pv_left * stored_to_ac
+
+
+def export_budget_update(mode, bucket, pv_part, grid_part, soc_before, soc_after, charge_kwh, pv_now, house_frac, export_kwh, cap):
+    """Advance the budget state over one step, returns (bucket, pv_part, grid_part).
+
+    charge_kwh is the DC energy drawn into the battery this step (0 when not charging), house_frac the
+    share of this step's battery output that served on-site load, export_kwh the energy that reached the
+    meter. pv_origin parts are renormalised to soc_after so they always sum to the SoC.
+    """
+    if mode == EXPORT_BUDGET_PV_EQUIVALENT:
+        bucket = min(max(bucket + pv_now - export_kwh, 0.0), cap)
+        return bucket, pv_part, grid_part
+    delta = soc_after - soc_before
+    if delta > 0:
+        pv_frac = min(charge_kwh, pv_now) / charge_kwh if charge_kwh > 0 else 0.0
+        pv_part += delta * pv_frac
+        grid_part += delta * (1.0 - pv_frac)
+    elif delta < 0:
+        removed = -delta
+        removed_house = removed * house_frac
+        removed_export = removed - removed_house
+        from_grid = min(grid_part, removed_house)
+        grid_part -= from_grid
+        pv_part = max(pv_part - (removed_house - from_grid), 0.0)
+        from_pv = min(pv_part, removed_export)
+        pv_part -= from_pv
+        grid_part = max(grid_part - (removed_export - from_pv), 0.0)
+    total = pv_part + grid_part
+    if total > 0:
+        scale = soc_after / total
+        pv_part *= scale
+        grid_part *= scale
+    else:
+        pv_part = 0.0
+        grid_part = soc_after
+    return bucket, pv_part, grid_part
+
+
+def export_budget_floor_percent(soc, exportable, soc_max):
+    """Lowest export target (whole percent, rounded up) that cannot sell more than `exportable` stored kWh from `soc`"""
+    if soc_max <= 0:
+        return 0
+    return min(max(int(math.ceil((soc - exportable) * 100.0 / soc_max - 1e-6)), 0), 100)
+
+
+def _replay_delta(data, ago, step):
+    """Energy in the step ending `ago` minutes ago, None for a counter glitch"""
+    value = data.get(ago, 0.0) - data.get(ago + step, 0.0)
+    if value < -0.05 or value > EXPORT_BUDGET_MAX_STEP_KWH:
+        return None
+    return max(value, 0.0)
+
+
+def _replay_soc_delta(soc_history, ago, step):
+    """Measured change of stored energy in the step ending `ago` minutes ago, None when unknown or a glitch"""
+    if ago not in soc_history or (ago + step) not in soc_history:
+        return None
+    value = soc_history[ago] - soc_history[ago + step]
+    if abs(value) > EXPORT_BUDGET_MAX_STEP_KWH:
+        return None
+    return value
+
+
+def export_budget_replay(mode, pv_today, export_today, import_today, load_minutes, soc_max, cap, minutes=EXPORT_BUDGET_REPLAY_DAYS * 24 * 60, step=5, soc_history=None):
+    """Replay measured history (cumulative series keyed by minutes ago) through the budget rules.
+
+    Returns (seed, noncompliant_kwh, steps). seed is (bucket, pv_share), or None when there is no usable
+    history. noncompliant_kwh is battery export the rules did not allow.
+
+    pv_origin labels battery flow per step. With soc_history (stored kWh keyed by minutes ago) the flow is the
+    measured SoC change, starting from the stored energy at the start of the window counted as grid; a drop the
+    house and the meter do not explain (losses, BMS resync) comes off the PV part first. Without it the flow is
+    estimated as PV + import - export - load, which reads conversion losses as charge, so it is only a fallback.
+    Only the PV share of the stored energy is kept; the prediction applies it to the real SoC.
+    """
+    series = [pv_today, export_today]
+    use_soc = False
+    if mode == EXPORT_BUDGET_PV_ORIGIN:
+        series += [import_today, load_minutes]
+        use_soc = bool(soc_history)
+        if use_soc:
+            series.append(soc_history)
+    if any(not data for data in series):
+        return None, 0.0, 0
+    available = min([minutes] + [max(data.keys()) for data in series])
+    available = (available // step) * step
+    bucket = 0.0
+    pv_part = 0.0
+    grid_part = soc_history.get(available, 0.0) if use_soc else 0.0
+    noncompliant = 0.0
+    steps = 0
+    for ago in range(available - step, -1, -step):
+        pv = _replay_delta(pv_today, ago, step)
+        exported = _replay_delta(export_today, ago, step)
+        if pv is None or exported is None:
+            continue
+        if mode == EXPORT_BUDGET_PV_EQUIVALENT:
+            noncompliant += max(exported - (bucket + pv), 0.0)
+            bucket = min(max(bucket + pv - exported, 0.0), cap)
+        else:
+            imported = _replay_delta(import_today, ago, step)
+            load = _replay_delta(load_minutes, ago, step)
+            if imported is None or load is None:
+                continue
+            if use_soc:
+                flow = _replay_soc_delta(soc_history, ago, step)
+                if flow is None:
+                    continue
+            else:
+                flow = pv + imported - exported - load
+            if flow > 0:
+                pv_in = min(flow, pv)
+                pv_part += pv_in
+                grid_part += flow - pv_in
+            elif flow < 0:
+                out = -flow
+                house = export_budget_house_from_battery(mode, out, load, pv)
+                to_grid = out - house
+                unexplained = 0.0
+                if use_soc:
+                    # The meter bounds what can have left for the grid; the rest is loss or a SoC resync
+                    unexplained = max(to_grid - exported, 0.0)
+                    to_grid -= unexplained
+                from_grid = min(grid_part, house)
+                grid_part -= from_grid
+                pv_part = max(pv_part - (house - from_grid), 0.0)
+                from_pv = min(pv_part, to_grid)
+                pv_part -= from_pv
+                excess = to_grid - from_pv
+                noncompliant += excess
+                grid_part = max(grid_part - excess, 0.0)
+                from_pv = min(pv_part, unexplained)
+                pv_part -= from_pv
+                grid_part = max(grid_part - (unexplained - from_pv), 0.0)
+            total = pv_part + grid_part
+            if soc_max > 0 and total > soc_max:
+                pv_part *= soc_max / total
+                grid_part *= soc_max / total
+        steps += 1
+    if steps == 0:
+        return None, 0.0, 0
+    if mode == EXPORT_BUDGET_PV_EQUIVALENT:
+        return (bucket, 0.0), noncompliant, steps
+    total = pv_part + grid_part
+    return (0.0, pv_part / total if total > 0 else 0.0), noncompliant, steps

apps/predbat/fetch.py

@@ -40,6 +40,7 @@
 from predbat_metrics import metrics
 from futurerate import FutureRate
 from axle import fetch_axle_sessions, load_axle_slot, fetch_axle_active
+from export_budget import EXPORT_BUDGET_OFF, EXPORT_BUDGET_PV_ORIGIN, EXPORT_BUDGET_MODE_NAMES, EXPORT_BUDGET_REPLAY_DAYS, parse_export_budget_mode, export_budget_replay
 
 import ast
 import copy
@@ -1259,6 +1260,9 @@ def fetch_sensor_data(self, save=True):
         self.high_export_rates = []
         self.octopus_slots = [[] for _ in range(self.num_cars)]
         self.cost_today_sofar = 0
+        self.export_budget_seed = None
+        self.export_budget_noncompliant_kwh = 0.0
+        self.export_budget_replay_steps = 0
         self.carbon_today_sofar = 0
         self.import_today = {}
         self.export_today = {}
@@ -1700,6 +1704,69 @@ def fetch_sensor_data(self, save=True):
             force_replan = True
         return force_replan
 
+    def update_export_budget_seed(self):
+        """Replay the measured history to find where the export energy budget stands now.
+
+        Runs after fetch_inverter_data() because the budget cap and the pv_origin parts need soc_max.
+        """
+        self.export_budget_seed = None
+        self.export_budget_noncompliant_kwh = 0.0
+        self.export_budget_replay_steps = 0
+        if not self.export_budget_mode:
+            return
+        cap = self.export_budget_cap_kwh if self.export_budget_cap_kwh > 0 else self.soc_max
+        soc_history = {}
+        if self.export_budget_mode == EXPORT_BUDGET_PV_ORIGIN:
+            soc_history = self.fetch_export_budget_soc_history()
+            if not soc_history:
+                self.log("Warn: Export energy budget has no battery SoC history - estimating the battery flow from pv/import/export/load, which counts conversion losses as PV charge")
+        seed, noncompliant, steps = export_budget_replay(self.export_budget_mode, self.pv_today, self.export_today, self.import_today, self.load_minutes, self.soc_max, cap, soc_history=soc_history)
+        if seed is None:
+            self.log("Warn: Export energy budget has no usable pv/export/import/load history - treating nothing as exportable")
+        self.export_budget_seed = seed
+        self.export_budget_noncompliant_kwh = noncompliant
+        self.export_budget_replay_steps = steps
+
+    def fetch_export_budget_soc_history(self):
+        """Stored kWh keyed by minutes ago over the export budget replay window, summed across inverters.
+
+        Read from each inverter's soc_percent (or soc_kw) sensor history. Returns {} when any inverter has none, so the
+        replay falls back to estimating the battery flow.
+        """
+        days = EXPORT_BUDGET_REPLAY_DAYS + 1
+        total = {}
+        counts = {}
+        for inverter in self.inverters:
+            soc_max = getattr(inverter, "soc_max", 0) or 0
+            percent_sensor = self.get_arg("soc_percent", indirect=False, index=inverter.id)
+            kw_sensor = self.get_arg("soc_kw", indirect=False, index=inverter.id)
+            sensor = percent_sensor or kw_sensor
+            if not isinstance(sensor, str) or not sensor or (percent_sensor and soc_max <= 0):
+                return {}
+            history = self.get_history_wrapper(entity_id=sensor, days=days, required=False)
+            if not history or not history[0]:
+                return {}
+            data, _ = minute_data(
+                history[0],
+                days,
+                self.now_utc,
+                "state",
+                "last_updated",
+                backwards=True,
+                clean_increment=False,
+                smoothing=False,
+                divide_by=1.0,
+                # soc_kw is scaled by battery_scaling, as the live SoC is (Inverter.update_status)
+                scale=soc_max / 100.0 if percent_sensor else (getattr(inverter, "battery_scaling", 1.0) or 1.0),
+                required_unit="%" if percent_sensor else "kWh",
+            )
+            if not data:
+                return {}
+            for minute, value in data.items():
+                total[minute] = total.get(minute, 0.0) + value
+                counts[minute] = counts.get(minute, 0) + 1
+        return {minute: value for minute, value in total.items() if counts[minute] == len(self.inverters)}
+
     def fetch_sensor_data_car_planning(self):
         # Work out car plan?
         # Determine which cars are configured with Octopus Intelligent to avoid calling plan_car_charging for them
@@ -3852,6 +3919,24 @@ def fetch_config_options(self):
         self.calculate_export_on_pv = self.get_arg("calculate_export_on_pv")
         self.calculate_second_pass = self.get_arg("calculate_second_pass")
         self.prediction_kernel_enable = self.get_arg("prediction_kernel_enable", True)
+        budget_arg = self.get_arg("export_budget_mode", "off")
+        budget_mode = parse_export_budget_mode(budget_arg)
+        if budget_mode is None:
+            self.log("Warn: export_budget_mode {} must be one of {} - export energy budget disabled".format(budget_arg, ", ".join(EXPORT_BUDGET_MODE_NAMES)))
+            budget_mode = EXPORT_BUDGET_OFF
+        self.export_budget_mode = budget_mode
+        cap_arg = self.get_arg("export_budget_cap_kwh", 0.0) or 0.0
+        try:
+            budget_cap = float(cap_arg)
+        except (TypeError, ValueError):
+            budget_cap = -1.0
+        if budget_cap < 0:
+            self.log("Warn: export_budget_cap_kwh {} must be a positive number of kWh - using the battery size".format(cap_arg))
+            budget_cap = 0.0
+        self.export_budget_cap_kwh = budget_cap
+        if budget_mode:
+            mode_name = [name for name, value in EXPORT_BUDGET_MODE_NAMES.items() if value == budget_mode][0]
+            self.log("Export energy budget enabled in {} mode".format(mode_name))
         self.calculate_inday_adjustment = self.get_arg("calculate_inday_adjustment")
         self.calculate_regions = True
         self.calculate_import_low_export = self.get_arg("calculate_import_low_export")