@@ -43,6 +43,16 @@
     # figure, without doing any arithmetic themselves.
     "quoted_pv_gbp": 0.0,
     "quoted_total_gbp": 0.0,
+    # Air source heat pump, installed. A typical 2025/26 UK retrofit including the
+    # cylinder and the radiator work an MCS installer signs off.
+    "heat_pump_install_gbp": 13000.0,
+    # The Boiler Upgrade Scheme grant, paid to the installer and deducted from the
+    # price the household actually pays, so it reduces CAPITAL rather than being an
+    # income stream. Zero it for a household that does not qualify.
+    "heat_pump_grant_gbp": 7500.0,
+    # A real quote for the heat pump installation, before the grant. Zero means "no
+    # quote" rather than "free", matching quoted_pv_gbp/quoted_total_gbp above.
+    "quoted_heat_pump_gbp": 0.0,
 }
 
 # Accepted in a stored config but ignored: an earlier revision asked for a battery-only
@@ -164,6 +174,55 @@ def build_costs(total_kwp, battery_kwh, settings):
     }
 
 
+def build_heat_costs(settings):
+    """Return the heat pump capital breakdown: list price, grant, and what is actually paid.
+
+    The grant reduces capital rather than appearing as a saving, because that is how the
+    Boiler Upgrade Scheme works in practice - it is paid to the installer and the
+    household is invoiced the difference. Treating it as an annual income stream would
+    make the payback period depend on how long it was spread over, which is meaningless.
+
+    The net figure is floored at zero: a grant larger than the price would otherwise
+    produce a negative capital cost and a nonsensical payback.
+    """
+    quoted = float(settings.get("quoted_heat_pump_gbp", 0) or 0)
+    install = quoted if quoted > 0 else float(settings.get("heat_pump_install_gbp", 0) or 0)
+    grant = min(float(settings.get("heat_pump_grant_gbp", 0) or 0), install)
+    return {
+        "install_gbp": round(install, 2),
+        "grant_gbp": round(grant, 2),
+        "net_gbp": round(max(0.0, install - grant), 2),
+        "quoted": quoted > 0,
+    }
+
+
+def build_heat_payback(heat_scenarios, heat_costs, months_included):
+    """Return the heat pump payback rows, or a reason they could not be built.
+
+    One row per hardware scenario, because the answer genuinely differs between them:
+    a heat pump's extra electricity is far cheaper for a household that can charge a
+    battery overnight and run the heat pump off it than for one importing every kWh at
+    the daytime rate. Collapsing that to a single number would hide the interaction this
+    tool exists to measure.
+
+    Like ``build_payback``, this refuses a partial year rather than extrapolating one:
+    heating is the most seasonal load in the house, so a year missing January is not
+    eleven twelfths of an answer - it is missing the part that matters most.
+    """
+    if not heat_scenarios:
+        return {"available": False, "reason": "No month produced a usable heat pump result, so there is nothing to pay back."}
+    if months_included != 12:
+        return {
+            "available": False,
+            "reason": "Heat pump payback needs a full year, but only {} of 12 months could be modelled. Heating is strongly seasonal, so a partial year cannot be scaled up. The missing months are named in the caveats.".format(months_included),
+        }
+
+    rows = {}
+    for key, entry in heat_scenarios.items():
+        rows[key] = payback_row(heat_costs["net_gbp"], entry["annual_saving_p"] / 100.0)
+    return {"available": True, "scenarios": rows}
+
+
 def payback_row(capital_gbp, annual_saving_gbp, recurring_gbp=0.0):
     """Return one payback row: capital divided by the net annual saving.
 
 
@@ -108,10 +108,20 @@ def __init__(self, config, log, predbat, storage=None, fetch_json=None, timezone
         self.unpaid_export_months = set()
 
     def _resolve_url(self, url, name, dno_region=None):
-        """Substitute templated arguments such as {dno_region} into a tariff URL, without mutating predbat.args."""
+        """Substitute templated arguments such as {dno_region} into a tariff URL, without mutating predbat.args.
+
+        Prefilled URLs can come from two different sources with two different
+        placeholder names for the same value: the tariff catalogue's own entries
+        use ``{dno_region}``, while URLs copied from the user's live apps.yaml
+        (``rates_import_octopus_url`` / ``rates_export_octopus_url`` - see
+        web_annual.py's prefill_config) use Predbat's own long-standing
+        ``{octopus_region}`` convention. Both must resolve to the same region
+        letter, so both keys are offered here regardless of which one the URL
+        actually contains.
+        """
         if not url:
             return None
-        extra_args = {"dno_region": dno_region} if dno_region else None
+        extra_args = {"dno_region": dno_region, "octopus_region": dno_region} if dno_region else None
         return self.predbat.resolve_arg(name, url, indirect=False, extra_args=extra_args)
 
     @staticmethod
 
@@ -13,7 +13,9 @@
 """
 
 import math
-from datetime import timedelta
+from datetime import datetime, timedelta
+
+import pytz
 
 from annual_http import fetch_json
 from solar_model import convert_azimuth, gti_hourly_to_period_kwh
@@ -23,6 +25,16 @@
 
 HOURLY_VARIABLES = "global_tilted_irradiance,temperature_2m,wind_speed_10m"
 
+# The heat model needs outdoor temperature and nothing else. Requested on its own rather
+# than lifted out of the PV download because a heat pump can be modelled with no solar
+# arrays configured at all, in which case there is no PV request to lift it from.
+TEMPERATURE_VARIABLE = "temperature_2m"
+
+# A local day must have at least this many of its 24 hours present to be usable. Short
+# days are a real occurrence at the edges of the downloaded window and on the spring
+# clock change; a day missing more than a few hours would understate its heat demand.
+MIN_HOURS_PER_LOCAL_DAY = 20
+
 # A month needs at least this many usable forecast/actual day pairs before its
 # measured P10 ratio is trusted over the flat fallback.
 MIN_DAYS_FOR_P10 = 7
@@ -230,6 +242,65 @@ def _derive_p10_ratios(self, actual_periods, forecast_periods, forecast_availabl
 
         return ratios, fallback_months
 
+    def _build_temperature_url(self, year):
+        """Build the Open-Meteo request for a calendar year of hourly outdoor temperature.
+
+        No tilt or azimuth: temperature is a property of the site, not of any array, so
+        this is a single request regardless of how many arrays are configured - or
+        whether any are.
+        """
+        return "{}?latitude={}&longitude={}&start_date={}-01-01&end_date={}-01-01&hourly={}&timezone=UTC".format(ARCHIVE_URL, self.latitude, self.longitude, year, year + 1, TEMPERATURE_VARIABLE)
+
+    @staticmethod
+    def _temperature_payload_problem(data):
+        """Return why a temperature payload cannot be trusted, or None when it is usable."""
+        if not data:
+            return "no data"
+        hourly = data.get("hourly", {})
+        times = hourly.get("time", [])
+        values = hourly.get(TEMPERATURE_VARIABLE, [])
+        if not times or not values:
+            return "no hourly values"
+        if len(values) < len(times):
+            return "truncated ({} stamps, {} values)".format(len(times), len(values))
+        return None
+
+    async def fetch_temperatures(self, year):
+        """Download a calendar year of hourly outdoor temperature, keyed by tz-aware UTC hour.
+
+        Returns an empty dict when the download is unusable, leaving the caller to decide
+        whether that is fatal - it is for a heat pump run, and irrelevant for one without.
+        Cached like the PV series, so a repeated run costs no request.
+        """
+        cache_key = "temperature_{}_{}_{}".format(year, self.latitude, self.longitude)
+        data = None
+        if self.storage:
+            cached = await self.storage.load("annual", cache_key)
+            if self._temperature_payload_problem(cached) is None:
+                data = cached
+        if not data:
+            fetched = await self.fetch_json(self._build_temperature_url(year))
+            problem = self._temperature_payload_problem(fetched)
+            if problem is not None:
+                self.log("Warn: Annual: outdoor temperature data for {} is unusable ({})".format(year, problem))
+                return {}
+            data = fetched
+            if self.storage:
+                await self.storage.save("annual", cache_key, data, format="json")
+
+        hourly = data["hourly"]
+        series = {}
+        for index, stamp_text in enumerate(hourly["time"]):
+            value = hourly[TEMPERATURE_VARIABLE][index]
+            if value is None:
+                continue
+            try:
+                stamp = datetime.strptime(stamp_text, "%Y-%m-%dT%H:%M").replace(tzinfo=pytz.utc)
+            except (ValueError, TypeError):
+                continue
+            series[stamp] = float(value)
+        return series
+
     async def fetch(self, year):
         """Download and convert a calendar year, returning a populated WeatherYear."""
         actual_periods = await self._fetch_series(ARCHIVE_URL, year, "actual")
@@ -243,6 +314,56 @@ async def fetch(self, year):
         return WeatherYear(actual_periods, forecast_periods, ratios, forecast_available, fallback_months)
 
 
+def temperature_by_local_day(series, timezone):
+    """Regroup a UTC-keyed hourly temperature series into 24 values per LOCAL date.
+
+    The heat model works in local days because the load profile does: minute 0 of a
+    ``load_forecast`` is local midnight, so hour 0 of a day's heat demand has to be
+    local midnight too. Feeding it a UTC-keyed series would shift the whole heating
+    profile by the UTC offset - an hour in British Summer Time, which lands the evening
+    peak in the wrong half-hourly rate slot.
+
+    Both clock changes are handled rather than discarded. In autumn two UTC hours map
+    onto the same local hour, so they are averaged; in spring one local hour never
+    happens, so it is filled from its neighbours along with any other isolated gap.
+    A local day still short of ``MIN_HOURS_PER_LOCAL_DAY`` real readings is dropped
+    entirely - that is the first and last day of the downloaded window, which only ever
+    hold a partial day, and a day filled mostly from guesses would understate its heat.
+    """
+    zone = pytz.timezone(timezone)
+    collected = {}
+    for stamp, value in series.items():
+        local = stamp.astimezone(zone)
+        collected.setdefault(local.date(), {}).setdefault(local.hour, []).append(value)
+
+    by_day = {}
+    for day, hours in collected.items():
+        if len(hours) < MIN_HOURS_PER_LOCAL_DAY:
+            continue
+        averaged = {hour: sum(values) / len(values) for hour, values in hours.items()}
+        by_day[day] = _fill_missing_hours(averaged)
+    return by_day
+
+
+def _fill_missing_hours(averaged):
+    """Return 24 hourly values, filling any gap from the nearest hour that has one.
+
+    Only ever called for a day already known to hold at least
+    ``MIN_HOURS_PER_LOCAL_DAY`` real readings, so a filled hour is always surrounded by
+    real ones - this is interpolation across a clock change or an isolated dropout, not
+    invention of a day that was never downloaded.
+    """
+    present = sorted(averaged)
+    filled = []
+    for hour in range(24):
+        if hour in averaged:
+            filled.append(averaged[hour])
+            continue
+        nearest = min(present, key=lambda candidate: abs(candidate - hour))
+        filled.append(averaged[nearest])
+    return filled
+
+
 POSTCODE_URL = "https://api.postcodes.io/postcodes/{}"
 
 
 
@@ -17,6 +17,7 @@
 
 from datetime import timedelta, datetime, timezone
 import asyncio
+import sqlite3
 import time
 import traceback
 import threading
@@ -26,6 +27,7 @@
 IPC_TIMEOUT = 60.0  # Seconds to wait for IPC response
 MAINTENANCE_HOUR = 3  # Local hour at which nightly database maintenance runs
 IDLE_WAIT = 30.0  # Seconds to wait for new work before re-checking housekeeping
+DB_ERROR_RECOVER_AFTER = 5  # Consecutive corruption errors before rebuilding the connection
 
 
 class DatabaseManager(ComponentBase):
@@ -50,6 +52,7 @@ def initialize(self, db_enable, db_days, db_days_daily=365):
         self.api_started = False
         self.last_success_timestamp = None
         self.last_commit_time = datetime.now(timezone.utc)
+        self.db_error_count = 0
 
     def bridge_event(self, loop):
         """
@@ -141,17 +144,50 @@ async def start(self):
                             self.db_engine._commit_db()
                             self.last_commit_time = now
                 self.last_success_timestamp = datetime.now(timezone.utc)
+                self.db_error_count = 0
 
             except Exception as e:
                 self.log(f"Error in database thread: {e}")
                 self.log("Error: " + traceback.format_exc())
+                self.recover_from_error(e)
 
         if hasattr(self.db_engine, "_commit_db"):
             self.db_engine._commit_db()
         self.db_engine._close()
         self.log("db_manager: Stopped cleanly")
         self.api_started = False
 
+    def recover_from_error(self, error):
+        """
+        Rebuild the connection after repeated corruption errors.
+
+        The queue loop otherwise has no recovery of its own - DatabaseEngine only retries
+        and quarantines at startup - so a single bad page leaves Predbat running with
+        db_primary set and every history write failing silently for the life of the process.
+        That is exactly what happened on 2026-08-02, and it only cleared by luck.
+
+        A lock or a busy file is not corruption: another reader will let go on its own and
+        reconnecting mid-contention would just make it worse, so those are left alone.
+        """
+        if not isinstance(error, sqlite3.DatabaseError) or isinstance(error, sqlite3.OperationalError):
+            return
+
+        self.db_error_count += 1
+        if self.db_error_count < DB_ERROR_RECOVER_AFTER:
+            return
+
+        self.log("Warn: db_manager: {} database errors in a row, rebuilding the connection".format(self.db_error_count))
+        self.db_error_count = 0
+        try:
+            # _cleanup_db_retry reconnects, re-checks, and quarantines the file if it really
+            # is damaged, so this recovers or starts a fresh database rather than looping
+            self.db_engine._connect()
+            self.db_engine._cleanup_db_retry()
+            self.log("Info: db_manager: Database connection rebuilt")
+        except Exception as e:
+            self.log("Error: db_manager: Unable to recover the database: {}".format(e))
+            self.log("Error: " + traceback.format_exc())
+
     def run_maintenance(self):
         """
         Run the nightly prune and vacuum once per day, in the small hours.
 
@@ -313,7 +313,10 @@ def fill_load_from_power(self, load_minutes, load_power_data):
 
         # Fill zero periods with integrated power data
         if zero_periods:
-            self.log("Warn: Found {} periods of zero load with power data, filling using power integration".format(len(zero_periods)))
+            # Not a warning - the gap is found and repaired here, so this is routine. A real
+            # historic outage would otherwise re-report itself every cycle until it aged out
+            # of the history window.
+            self.log("Info: Found {} periods of zero load with power data, filling using power integration".format(len(zero_periods)))
             # Print the first 5 periods for debugging
             for i, (period_start, period_end, base_value) in enumerate(zero_periods[:5]):
                 start_timestamp = self.now_utc - timedelta(minutes=period_start)