perf(ha): fetch long history windows in chunks to bound peak memory

Predbat fetches up to 22 days of history for several sensors on its first plan
cycle. requests computes .json() as loads(self.text), and .text decodes
.content, so during the parse the same payload is resident three times over -
raw bytes, decoded string and parsed objects. For load_power that is 16.3MB of
JSON and 54,000 states, and profiling showed those three representations
dominating the peak memory of a run.

Split windows longer than HISTORY_CHUNK_DAYS across several requests so only one
chunk's three representations are live at a time. get_history_window() is
extracted for the single request and get_history() assembles the chunks.

Home Assistant opens every history window with the state in effect at
start_time, synthesised rather than recorded. For chunks after the first that
instant is already covered by the previous chunk, so the record is a duplicate.
It cannot simply be ignored as harmless: when a boundary lands inside a gap in
the recording it adds a data point a single request never returns, and
minute_data then interpolates across that gap differently. Against 21 days of
real history one boundary in six did exactly that, moving 12 minutes of load
data. Dropping records at or before each chunk's start makes the state count
match a single fetch exactly and the resulting minute data byte-identical, at
every chunk size tested.

Measured over three plan cycles against a live Home Assistant, 22 day window:

  chunk size   peak RSS   settled RSS   cold cycle
  none          221.3MB       186.8MB        9.84s
  3 days        188.7MB       173.1MB        9.43s
  1 day         176.4MB       172.3MB       11.61s

Three days is the default: it takes essentially all the steady-state saving and
three quarters of the peak saving at no measurable time cost, where one day adds
1.8s of startup for 7 requests per sensor rather than 21. Warm cycles are
unaffected either way - the history cache refreshes with a small delta fetch
that stays under the chunk threshold.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: springfall2008

apps/predbat/ha.py

@@ -32,6 +32,10 @@
 from const import TIME_FORMAT_HA, TIMEOUT, TIME_FORMAT_HA_TZ
 from component_base import ComponentBase
 
+# Maximum days of history fetched per request. Long windows are split so only one chunk's
+# response body, decoded string and parsed objects are resident at a time.
+HISTORY_CHUNK_DAYS = 3
+
 
 class RunThread(threading.Thread):
     def __init__(self, coro):
@@ -869,12 +873,36 @@ def update_states(self):
         else:
             self.log("Warn: Failed to update state data from HA")
 
-    def get_history(self, sensor, now, days=30, from_time=None, force_db=False):
+    def get_history_window(self, sensor, start, end):
+        """
+        Fetch a single window of history for a sensor.
+
+        :param sensor: The sensor to get the history for.
+        :param start: Start of the window.
+        :param end: End of the window.
+        :return: The raw API response, or None.
+        """
+        res = self.api_call("/api/history/period/{}".format(start.strftime(TIME_FORMAT_HA)), {"filter_entity_id": sensor, "end_time": end.strftime(TIME_FORMAT_HA)})
+        if isinstance(res, list) and len(res) > 0:
+            return res
+        return None
+
+    def get_history(self, sensor, now, days=30, from_time=None, force_db=False, chunk_days=HISTORY_CHUNK_DAYS):
         """
         Get the history for a sensor from Home Assistant.
 
+        Long windows are fetched in chunks so only one chunk's response body, decoded string
+        and parsed objects are resident at a time rather than the whole window's. A 21 day
+        window of a power sensor is tens of megabytes of JSON, and holding all three
+        representations of it at once dominated the peak memory of a plan cycle.
+
         :param sensor: The sensor to get the history for.
-        :return: The history for the sensor.
+        :param now: Current time, the end of the window.
+        :param days: How many days of history to fetch.
+        :param from_time: Explicit window start, overriding days.
+        :param force_db: Read from the database rather than Home Assistant.
+        :param chunk_days: Maximum days per request; 0 or None fetches the window in one request.
+        :return: The history for the sensor, oldest first, or None.
         """
         if not sensor:
             return None
@@ -889,11 +917,28 @@ def get_history(self, sensor, now, days=30, from_time=None, force_db=False):
         else:
             start = now - timedelta(days=days)
         end = now
-        res = self.api_call("/api/history/period/{}".format(start.strftime(TIME_FORMAT_HA)), {"filter_entity_id": sensor, "end_time": end.strftime(TIME_FORMAT_HA)})
-        if isinstance(res, list) and len(res) > 0:
-            return res
-        else:
-            return None
+
+        if not chunk_days or (end - start) <= timedelta(days=chunk_days):
+            return self.get_history_window(sensor, start, end)
+
+        history = []
+        cursor = start
+        while cursor < end:
+            window_end = min(cursor + timedelta(days=chunk_days), end)
+            res = self.get_history_window(sensor, cursor, window_end)
+            if res:
+                for item in res[0]:
+                    # Home Assistant opens every window with the state in effect at start_time.
+                    # For chunks after the first that instant is already covered by the previous
+                    # chunk, and the synthesised record can land inside a gap in the recording
+                    # where it would add a data point a single request never returns, changing
+                    # how minute_data interpolates across that gap.
+                    if cursor > start and item.get("last_updated") and str2time(item["last_updated"]) <= cursor:
+                        continue
+                    history.append(item)
+            cursor = window_end
+
+        return [history] if history else None
 
     async def set_state_external(self, entity_id, state, attributes={}):
         """

apps/predbat/tests/test_history_chunking.py

@@ -0,0 +1,172 @@
+# fmt: off
+"""
+History Chunking Tests
+
+Tests for fetching long history windows from Home Assistant in time-ordered chunks
+rather than one request, so only one chunk's response is resident at a time:
+- get_history() splits long windows and covers them exactly
+- the synthesised record Home Assistant returns at each chunk start is dropped
+"""
+
+from datetime import datetime, timedelta, timezone
+
+from tests.test_hainterface_common import MockBase, create_ha_interface
+
+
+NOW = datetime(2026, 8, 16, 12, 0, 0, tzinfo=timezone.utc)
+
+
+def _interface():
+    """Build an HAInterface against the shared mock base."""
+    return create_ha_interface(MockBase(), ha_key="test_key", db_enable=False, db_mirror_ha=False, db_primary=False)
+
+
+def _record(stamp, state):
+    """Build one history record."""
+    return {"state": state, "last_updated": stamp.isoformat(), "attributes": {"unit_of_measurement": "W"}}
+
+
+def _capture(interface, responder):
+    """Point api_call at a responder and record the windows requested."""
+    calls = []
+
+    def api_call(endpoint, data_in=None, **kwargs):
+        """Stand in for the REST call, recording the requested window."""
+        start = endpoint.rsplit("/", 1)[1]
+        calls.append((start, data_in.get("end_time")))
+        return responder(start, data_in.get("end_time"))
+
+    interface.api_call = api_call
+    return calls
+
+
+def test_get_history_drops_synthesised_boundary_record(my_predbat=None):
+    """Home Assistant's state-at-window-start record must not be added twice"""
+    print("\n=== Testing get_history() boundary record handling ===")
+    failed = 0
+
+    interface = _interface()
+
+    def responder(start, end):
+        """Return a synthesised record at the window start plus one real record."""
+        start_dt = datetime.fromisoformat(start)
+        # Home Assistant reports the state in effect at start_time, exactly on the boundary
+        # and with no sub-second precision, then any real changes within the window
+        return [[_record(start_dt, "500"), _record(start_dt + timedelta(hours=1), "600")]]
+
+    _capture(interface, responder)
+
+    result = interface.get_history("sensor.load_power", NOW, days=9, chunk_days=3)
+
+    if not result or not result[0]:
+        print("ERROR: expected history back, got {}".format(result))
+        return failed + 1
+
+    items = result[0]
+    stamps = [item["last_updated"] for item in items]
+    if len(stamps) != len(set(stamps)):
+        duplicates = sorted({s for s in stamps if stamps.count(s) > 1})
+        print("ERROR: chunk boundaries duplicated records: {}".format(duplicates))
+        failed += 1
+    else:
+        print("✓ no duplicate timestamps across chunk boundaries")
+
+    # 3 chunks: chunk 1 contributes both records, chunks 2 and 3 contribute only their real one
+    if len(items) != 4:
+        print("ERROR: expected 4 records (2 from the first chunk, 1 from each later chunk), got {}".format(len(items)))
+        print("       records: {}".format(stamps))
+        failed += 1
+    else:
+        print("✓ synthesised boundary record dropped for chunks after the first")
+
+    if stamps != sorted(stamps):
+        print("ERROR: records not in oldest-first order: {}".format(stamps))
+        failed += 1
+    else:
+        print("✓ records stay oldest-first across chunks")
+
+    return failed
+
+
+def test_get_history_short_window_makes_one_request(my_predbat=None):
+    """A window inside the chunk size is fetched in a single request as before"""
+    print("\n=== Testing get_history() short window ===")
+    failed = 0
+
+    interface = _interface()
+    calls = _capture(interface, lambda start, end: [[_record(NOW - timedelta(hours=1), "500")]])
+
+    interface.get_history("sensor.load_power", NOW, days=2, chunk_days=7)
+
+    if len(calls) != 1:
+        print("ERROR: expected 1 request for a 2 day window, got {}".format(len(calls)))
+        failed += 1
+    else:
+        print("✓ short window still fetched in one request")
+
+    return failed
+
+
+def test_get_history_chunks_cover_the_window_exactly(my_predbat=None):
+    """Chunks tile the requested window end to end with no gaps or overlap"""
+    print("\n=== Testing get_history() window coverage ===")
+    failed = 0
+
+    interface = _interface()
+    calls = _capture(interface, lambda start, end: [[_record(datetime.fromisoformat(start) + timedelta(hours=1), "500")]])
+
+    interface.get_history("sensor.load_power", NOW, days=10, chunk_days=3)
+
+    if len(calls) != 4:
+        print("ERROR: expected 4 chunks for 10 days at 3 days each, got {}".format(len(calls)))
+        return failed + 1
+    print("✓ 10 day window split into 4 chunks of at most 3 days")
+
+    starts = [datetime.fromisoformat(start) for start, _ in calls]
+    ends = [datetime.fromisoformat(end) for _, end in calls]
+
+    if starts[0] != NOW - timedelta(days=10):
+        print("ERROR: first chunk starts at {}, expected {}".format(starts[0], NOW - timedelta(days=10)))
+        failed += 1
+    elif ends[-1] != NOW:
+        print("ERROR: last chunk ends at {}, expected {}".format(ends[-1], NOW))
+        failed += 1
+    else:
+        print("✓ chunks span exactly the requested window")
+
+    for index in range(1, len(starts)):
+        if starts[index] != ends[index - 1]:
+            print("ERROR: chunk {} starts at {} but previous ended at {}".format(index, starts[index], ends[index - 1]))
+            failed += 1
+            break
+    else:
+        print("✓ chunks are contiguous with no gaps or overlap")
+
+    if max((ends[i] - starts[i]) for i in range(len(starts))) > timedelta(days=3):
+        print("ERROR: a chunk exceeded the 3 day limit")
+        failed += 1
+    else:
+        print("✓ no chunk exceeds the requested chunk size")
+
+    return failed
+
+
+def run_history_chunking_tests(my_predbat=None):
+    """Run all history chunking tests"""
+    print("\n" + "=" * 80)
+    print("History Chunking Tests")
+    print("=" * 80)
+
+    failed = 0
+    failed += test_get_history_drops_synthesised_boundary_record(my_predbat)
+    failed += test_get_history_short_window_makes_one_request(my_predbat)
+    failed += test_get_history_chunks_cover_the_window_exactly(my_predbat)
+
+    print("\n" + "=" * 80)
+    if failed == 0:
+        print("✅ All history chunking tests passed!")
+    else:
+        print(f"❌ {failed} history chunking test(s) failed")
+    print("=" * 80)
+
+    return failed

apps/predbat/unit_test.py

@@ -78,6 +78,7 @@
 from tests.test_hainterface_service import run_hainterface_service_tests
 from tests.test_hainterface_lifecycle import run_hainterface_lifecycle_tests
 from tests.test_hainterface_websocket import run_hainterface_websocket_tests
+from tests.test_history_chunking import run_history_chunking_tests
 from tests.test_web_if import run_test_web_if
 from tests.test_web_chart_currency import test_rates_chart_series_names_use_currency_symbol
 from tests.test_metrics_dashboard_soc_refresh import test_soc_chart_center_text_reads_live_data
@@ -489,6 +490,8 @@ def main():
         ("hainterface_lifecycle", run_hainterface_lifecycle_tests, "HAInterface lifecycle tests", False),
         # HAInterface websocket tests
         ("hainterface_websocket", run_hainterface_websocket_tests, "HAInterface websocket tests", False),
+        # History chunking (long windows fetched in pieces) tests
+        ("history_chunking", run_history_chunking_tests, "History chunking tests", False),
         # Carbon Intensity API unit tests
         ("carbon", test_carbon, "Carbon Intensity API comprehensive tests (fetch, cache, publish, config)", False),
         # Storage component unit tests