fix(data): chunk backfill's candle requests too, the last unpaged window site

keel/data/market_feed.py

@@ -58,6 +58,25 @@ def _align_up(ts: int, gran_sec: int) -> int:
     return ((ts + gran_sec - 1) // gran_sec) * gran_sec
 
 
+def _request_windows(start: int, end: int, gran_sec: int) -> list[tuple[int, int]]:
+    """Tile the inclusive `[start, end]` range into windows under the venue's candle cap.
+
+    One definition of the arithmetic for every candle request this module makes. The `- 1`
+    keeps each inclusive `[window_start, window_end]` at exactly `MAX_CANDLES_PER_REQUEST`
+    candles -- an inclusive range of `+ step` would be one over, which is the off-by-one #271
+    had to correct in `history.py`, and having it written twice here is how that happens.
+
+    Returns a single window when the range already fits, so a small gap still costs one request.
+    """
+    windows: list[tuple[int, int]] = []
+    window_start = start
+    while window_start <= end:
+        window_end = min(end, window_start + (MAX_CANDLES_PER_REQUEST - 1) * gran_sec)
+        windows.append((window_start, window_end))
+        window_start = window_end + gran_sec
+    return windows
+
+
 def _missing_ranges(expected: list[int], present: set[int], gran_sec: int) -> list[tuple[int, int]]:
     """Group the `expected` ts values not in `present` into contiguous `(start, end)` ranges."""
     missing = [ts for ts in expected if ts not in present]
@@ -112,14 +131,18 @@ def backfill(
             }
 
             for range_start, range_end in _missing_ranges(expected, existing, gran_sec):
-                fetched = client.get_candles(product_id, granularity, range_start, range_end)
-                gap_candles = [
-                    c
-                    for c in fetched
-                    if window_start <= c.ts <= latest_closed and c.ts not in existing
-                ]
-                if gap_candles:
-                    total_written += repo.upsert_candles(product_id, granularity, gap_candles)
+                # A contiguous missing range is itself unbounded -- an empty repo makes the
+                # whole history window one range -- so page it under the venue's candle cap.
+                # Upserted per window, so a mid-range failure leaves earlier windows persisted.
+                for req_start, req_end in _request_windows(range_start, range_end, gran_sec):
+                    fetched = client.get_candles(product_id, granularity, req_start, req_end)
+                    gap_candles = [
+                        c
+                        for c in fetched
+                        if window_start <= c.ts <= latest_closed and c.ts not in existing
+                    ]
+                    if gap_candles:
+                        total_written += repo.upsert_candles(product_id, granularity, gap_candles)
 
     return total_written
 
@@ -137,19 +160,14 @@ def _poll_catch_up(
     """Fetch and upsert `[fetch_start, latest_closed]`, chunked under the venue's candle cap.
 
     Mirrors `history._fill_forward`'s windowing idiom: page forward in windows of at most
-    `MAX_CANDLES_PER_REQUEST` candles each, upserting per window for incremental durability
-    (if a later window raises, earlier windows are already persisted). An empty window does
-    *not* stop the loop -- a mid-history hole must not block catch-up of newer candles.
+    `MAX_CANDLES_PER_REQUEST` candles each (see `_request_windows`), upserting per window for
+    incremental durability (if a later window raises, earlier windows are already persisted).
+    An empty window does *not* stop the loop -- a mid-history hole must not block catch-up of
+    newer candles.
     """
     total_written = 0
     seen: set[int] = set()
-    window_start = fetch_start
-    while window_start <= latest_closed:
-        # `- 1` keeps each inclusive [window_start, window_end] range at exactly
-        # MAX_CANDLES_PER_REQUEST candles (an inclusive range of `+ step` would be one over).
-        window_end = min(
-            latest_closed, window_start + (MAX_CANDLES_PER_REQUEST - 1) * gran_sec
-        )
+    for window_start, window_end in _request_windows(fetch_start, latest_closed, gran_sec):
         fetched = client.get_candles(product_id, granularity, window_start, window_end)
         new_candles: list[Candle] = [
             c
@@ -159,7 +177,6 @@ def _poll_catch_up(
         if new_candles:
             seen.update(c.ts for c in new_candles)
             total_written += repo.upsert_candles(product_id, granularity, new_candles)
-        window_start = window_end + gran_sec
 
     return total_written
 

tests/data/test_market_feed.py

@@ -316,3 +316,69 @@ def test_is_fresh_false_when_no_candles_stored(repo):
     assert not is_fresh(
         repo, "BTC-USD", Granularity.ONE_HOUR, now_ts=NOW, max_age_sec=200
     )
+
+
+# -- backfill: the same candle-cap defect, on the one windowing site #269/#271 did not reach ----
+#
+# #269 chunked `poll_once` and #271 chunked `repair.py` and `history.py`. `backfill` groups
+# missing timestamps into CONTIGUOUS ranges via `_missing_ranges` and asked for each range in a
+# single request -- so a contiguous hole wider than the cap 400s exactly as the poll path did.
+# Latent today (no production caller; `keel fetch` goes through `history.ensure_history`), but
+# it is the same defect class, and #271's stated goal was that every candle-request windowing
+# site in the codebase agree.
+
+BACKFILL_HOURS = 552  # same span as the real ZEC-USD gap, comfortably over the cap
+BACKFILL_DAYS = BACKFILL_HOURS // 24 + 1
+_BACKFILL_RAW_START = NOW - BACKFILL_DAYS * 86400
+# `backfill` aligns its window start UP to the next granularity boundary; mirrored here rather
+# than importing the private helper, so the test pins the observable behaviour.
+BACKFILL_WINDOW_START = ((_BACKFILL_RAW_START + GRAN_SEC - 1) // GRAN_SEC) * GRAN_SEC
+BACKFILL_TS = list(range(BACKFILL_WINDOW_START, LATEST_CLOSED + 1, GRAN_SEC))
+
+
+def _wide_series(product_id: str = "BTC-USD") -> dict[tuple[str, Granularity], list[Candle]]:
+    return {(product_id, Granularity.ONE_HOUR): [_candle(ts) for ts in BACKFILL_TS + [NOW]]}
+
+
+def test_backfill_never_requests_more_than_the_candle_cap(repo):
+    """An empty repo makes the whole history window one contiguous missing range."""
+    client = FakeClient(_wide_series())
+
+    backfill(
+        client, repo, ["BTC-USD"], [Granularity.ONE_HOUR],
+        history_days=BACKFILL_DAYS, now_ts=NOW,
+    )
+
+    assert client.calls
+    for _, _, start, end in client.calls:
+        assert start <= end
+        candle_count = (end - start) // GRAN_SEC + 1
+        assert candle_count <= MAX_CANDLES_PER_REQUEST
+
+
+def test_backfill_chunk_windows_are_contiguous_and_cover_the_gap(repo):
+    client = FakeClient(_wide_series())
+
+    written = backfill(
+        client, repo, ["BTC-USD"], [Granularity.ONE_HOUR],
+        history_days=BACKFILL_DAYS, now_ts=NOW,
+    )
+
+    assert len(client.calls) > 1, "a range this wide can only tile into >1 window under the cap"
+    assert client.calls[0][2] == BACKFILL_WINDOW_START
+    assert client.calls[-1][3] == LATEST_CLOSED
+    for previous, current in zip(client.calls, client.calls[1:]):
+        assert current[2] == previous[3] + GRAN_SEC
+    assert written == len(BACKFILL_TS), "every closed candle in the window should be persisted"
+
+
+def test_backfill_still_uses_one_request_for_a_gap_within_the_cap(repo):
+    """Regression guard: chunking must not add requests to the ordinary small-window case."""
+    client = FakeClient(_full_series())
+
+    backfill(
+        client, repo, ["BTC-USD"], [Granularity.ONE_HOUR],
+        history_days=HISTORY_DAYS, now_ts=NOW,
+    )
+
+    assert len(client.calls) == 1