fix(data): chunk the remaining candle-request windows under the venue cap

keel/data/history.py

@@ -111,7 +111,9 @@ def _fill_forward(
     """Fetch any bars strictly newer than `latest_cached`, up to `now_ts` (recent-bar gaps)."""
     window_start = latest_cached + step
     while window_start <= now_ts:
-        window_end = min(now_ts, window_start + MAX_CANDLES_PER_REQUEST * step)
+        # Inclusive range: [a, a + N*step] spans N+1 candles, so the cap needs the -1 or a
+        # request for MAX_CANDLES_PER_REQUEST candles actually asks for one more than that.
+        window_end = min(now_ts, window_start + (MAX_CANDLES_PER_REQUEST - 1) * step)
         batch = client.get_candles(product, granularity, window_start, window_end)
         if batch:
             repo.upsert_candles(product, granularity, batch)
@@ -133,7 +135,8 @@ def _fill_backward(
     """Page backward from `window_end` down to `start_floor`, stopping at the first empty
     window -- that window is either the asset's inception or already-covered territory."""
     while window_end >= start_floor:
-        window_start = max(start_floor, window_end - MAX_CANDLES_PER_REQUEST * step)
+        # Same inclusive-range arithmetic as `_fill_forward`: N*step would span N+1 candles.
+        window_start = max(start_floor, window_end - (MAX_CANDLES_PER_REQUEST - 1) * step)
         batch = client.get_candles(product, granularity, window_start, window_end)
         if not batch:
             break  # inception (or a confirmed-empty probe) -- nothing older to fetch

keel/data/repair.py

@@ -14,6 +14,21 @@
 ⚠️ That record is an ASSERTION ABOUT AN OBSERVATION ("we asked and it had nothing"), not an
 assumption. It is only ever written after an actual fetch attempt, never inferred -- which is
 why the v5 migration deliberately backfills nothing.
+
+**A gap window can be arbitrarily large.** `gaps.detect` puts no cap on `n_missing` -- an
+interior hole just accumulates for as long as the venue was unreachable or the asset was thin.
+Coinbase rejects any single request spanning more than ~350 candles, so a large-enough hole
+requested in one call 400s. Left unchunked, that failure is swallowed into `result.errors` and
+the pass moves on -- which means the same oversized request gets retried, and 400s again, on
+every single scheduled run forever, and *silently* since a logged error is easy to miss. So the
+widened window is paged in `MAX_CANDLES_PER_REQUEST`-sized chunks, each upserted as it arrives:
+a fetch that fails partway through still leaves the earlier chunks persisted, and the next run
+resumes further along instead of repeating the whole doomed request.
+
+That chunking is PREVENTATIVE. No cached series is anywhere near the cap today -- the largest
+interior gap across the deployment is 15 bars -- but a multi-day venue outage, a delisting and
+relisting, or a product added back after a long absence would each clear ~349 in a single hole,
+and the trap only announces itself as a series that quietly never repairs.
 """
 
 from __future__ import annotations
@@ -22,7 +37,7 @@
 from dataclasses import dataclass, field
 
 from keel.data import gaps as gaps_mod
-from keel.data.history import GRANULARITY_SECONDS
+from keel.data.history import GRANULARITY_SECONDS, MAX_CANDLES_PER_REQUEST
 from keel.types import Granularity
 
 
@@ -46,6 +61,36 @@ def bars_still_missing(self) -> int:
     remaining: list[gaps_mod.GapWindow] = field(default_factory=list)
 
 
+def _fetch_window_chunked(
+    client,
+    repo,
+    product: str,
+    granularity: Granularity,
+    window: gaps_mod.GapWindow,
+    step: int,
+    sleep_fn,
+    sleep_sec: float,
+) -> None:
+    """Fetch one gap window's widened range, one `MAX_CANDLES_PER_REQUEST`-sized chunk at a
+    time, upserting each chunk as it arrives.
+
+    The ±`step` widening (venues are inconsistent about endpoint inclusivity) applies only to
+    these OUTER edges -- internal chunk boundaries are contiguous, not themselves widened.
+    Upserting per chunk, rather than batching the whole window, is what lets an interior chunk
+    failure leave the earlier chunks persisted instead of losing the whole fetch.
+    """
+    fetch_start = window.start_ts - step
+    fetch_end = window.end_ts + step
+    chunk_start = fetch_start
+    while chunk_start <= fetch_end:
+        chunk_end = min(fetch_end, chunk_start + (MAX_CANDLES_PER_REQUEST - 1) * step)
+        fetched = client.get_candles(product, granularity, chunk_start, chunk_end)
+        if fetched:
+            repo.upsert_candles(product, granularity, fetched)
+        sleep_fn(sleep_sec)
+        chunk_start = chunk_end + step
+
+
 def repair_series(
     client,
     repo,
@@ -84,21 +129,20 @@ def repair_series(
         result.windows_probed += 1
         try:
             # Widen by one step each side: venues are inconsistent about endpoint inclusivity,
-            # and over-asking costs nothing because `upsert_candles` is idempotent.
-            fetched = client.get_candles(
-                product, granularity, window.start_ts - step, window.end_ts + step
+            # and over-asking costs nothing because `upsert_candles` is idempotent. The window
+            # itself may be far larger than one request can hold, so this pages internally.
+            _fetch_window_chunked(
+                client, repo, product, granularity, window, step, sleep_fn, sleep_sec
             )
         except Exception as exc:  # noqa: BLE001 -- one bad window must not abort the pass
             # NOT recorded as absent: a fetch that never completed proves nothing about
-            # whether the venue holds the data.
+            # whether the venue holds the data. Chunks that DID complete before the failure
+            # were already upserted, so a later run resumes past them rather than restarting.
             result.errors.append(f"{window.start_ts}-{window.end_ts}: {exc}")
             continue
 
-        # Only a COMPLETED request can testify that the venue lacks the data.
+        # Only a window whose EVERY chunk completed can testify that the venue lacks the data.
         probed_ok.append(window)
-        if fetched:
-            repo.upsert_candles(product, granularity, fetched)
-        sleep_fn(sleep_sec)
 
     after = repo.get_candles(product, granularity)
     result.bars_recovered = max(0, len(after) - len(before))

tests/data/test_gap_repair.py

@@ -10,6 +10,7 @@
 from keel.data import gaps as gaps_mod
 from keel.data import repair as repair_mod
 from keel.data.db import connect, migrate
+from keel.data.history import MAX_CANDLES_PER_REQUEST
 from keel.data.repository import Repository
 from keel.types import Candle, Granularity
 
@@ -129,6 +130,26 @@ def _seed(repo, missing: set[int], n: int = 10, product="BTC-USD"):
     )
 
 
+class _CappedVenue(_Venue):
+    """Like `_Venue`, but actually enforces Coinbase's real per-request ceiling.
+
+    `_Venue` alone never rejects an oversized ask, so it can't tell an unchunked repair apart
+    from a chunked one. This is what turns "the request would 400 in production" into something
+    a test can observe: any single call spanning >349 candles blows up, same as the real venue.
+    """
+
+    _CAP = 349  # Coinbase's real ceiling; MAX_CANDLES_PER_REQUEST (300) must stay under it
+
+    def get_candles(self, product, granularity, start, end):
+        self.calls.append((start, end))
+        n_requested = (end - start) // _DAY + 1
+        if n_requested > self._CAP:
+            raise RuntimeError(
+                "400 INVALID_ARGUMENT: number of candles requested should be less than 350"
+            )
+        return [_candle(ts) for ts in sorted(self.available) if start <= ts <= end]
+
+
 def test_repair_recovers_a_hole_the_venue_has(repo):
     _seed(repo, missing={4, 5})
     venue = _Venue({_BASE + i * _DAY for i in range(10)})
@@ -254,3 +275,138 @@ def test_a_shifted_window_is_treated_as_new_and_re_probed(repo):
     third = repair_mod.repair_series(venue, repo, "BTC-USD", Granularity.ONE_DAY, now_ts=3)
     assert third.windows_skipped_known_absent == 0
     assert third.windows_probed == 1
+
+
+# -- chunking a gap window that exceeds the per-request cap -------------------
+#
+# PREVENTATIVE, not a reproduction: no cached series has an interior gap anywhere near the cap
+# today (the largest across the deployment is 15 bars, against a ~349 limit). But an interior
+# gap that big is entirely reachable -- a multi-day venue outage, a delisting-and-relisting, a
+# product added back after a long absence -- and the failure mode if it ever happens is the bad
+# kind: one request for the whole window 400s ("number of candles requested should be less than
+# 350"), `repair_series` swallows it into `result.errors` and continues, and every scheduled run
+# thereafter retries the same doomed request. Silent and permanent. These tests pin the fix:
+# page the widened outer range in contiguous, non-overlapping chunks of at most
+# `MAX_CANDLES_PER_REQUEST` candles. 553 is used as the oversized-gap figure throughout.
+
+
+def test_a_large_hole_is_fetched_in_capped_chunks(repo):
+    """The widened outer range spans 555 candles (553 missing + one step on each side), so it
+    must page as 300 then 255 -- never one request the venue would reject."""
+    n_missing = 553
+    _seed(repo, missing=set(range(1, n_missing + 1)), n=n_missing + 2)
+    venue = _CappedVenue({_BASE + i * _DAY for i in range(n_missing + 2)})
+
+    repair_mod.repair_series(venue, repo, "BTC-USD", Granularity.ONE_DAY, now_ts=999)
+
+    assert len(venue.calls) == 2
+    counts = [(end - start) // _DAY + 1 for start, end in venue.calls]
+    assert counts == [300, 255]
+    assert all(count <= MAX_CANDLES_PER_REQUEST for count in counts)
+
+
+def test_a_large_hole_is_fully_recovered_with_no_duplicates_or_drops(repo):
+    n_missing = 553
+    _seed(repo, missing=set(range(1, n_missing + 1)), n=n_missing + 2)
+    venue = _CappedVenue({_BASE + i * _DAY for i in range(n_missing + 2)})
+
+    result = repair_mod.repair_series(venue, repo, "BTC-USD", Granularity.ONE_DAY, now_ts=999)
+
+    assert result.bars_recovered == n_missing
+    assert result.remaining == []
+    stored = repo.get_candles("BTC-USD", Granularity.ONE_DAY)
+    ts_values = [c.ts for c in stored]
+    assert ts_values == sorted(set(ts_values))  # ascending, no duplicates
+    assert len(stored) == n_missing + 2  # nothing dropped either
+
+
+def test_chunk_windows_tile_the_outer_range_without_widening_interior_boundaries(repo):
+    """The ±step widening is load-bearing only at the OUTER edges (venues disagree about
+    endpoint inclusivity); internal chunk boundaries must stay contiguous, not overlap, and
+    must not themselves be widened."""
+    n_missing = 553
+    _seed(repo, missing=set(range(1, n_missing + 1)), n=n_missing + 2)
+    venue = _CappedVenue({_BASE + i * _DAY for i in range(n_missing + 2)})
+
+    repair_mod.repair_series(venue, repo, "BTC-USD", Granularity.ONE_DAY, now_ts=999)
+
+    counts = [(end - start) // _DAY + 1 for start, end in venue.calls]
+    assert all(count <= MAX_CANDLES_PER_REQUEST for count in counts)
+    assert venue.calls[0][0] == _BASE + 0 * _DAY  # window.start_ts - step
+    assert venue.calls[-1][1] == _BASE + (n_missing + 1) * _DAY  # window.end_ts + step
+    for (_, prev_end), (next_start, _) in zip(venue.calls, venue.calls[1:]):
+        assert next_start == prev_end + _DAY
+
+
+def test_one_bad_chunk_is_not_recorded_absent_and_a_later_window_still_repairs(repo):
+    """A partially-fetched window proves nothing about whether the venue holds the rest of it --
+    it must not be recorded absent, and a later, separate gap in the same pass must still get
+    probed and filled rather than the whole pass aborting."""
+
+    class _FlakySecondChunk(_CappedVenue):
+        def get_candles(self, product, granularity, start, end):
+            if start == _BASE + 300 * _DAY:  # the big window's second chunk
+                raise RuntimeError("boom")
+            return super().get_candles(product, granularity, start, end)
+
+    n_missing = 553
+    missing = set(range(1, n_missing + 1)) | {700, 701}
+    _seed(repo, missing=missing, n=706)
+    venue = _FlakySecondChunk({_BASE + i * _DAY for i in range(706)})
+
+    result = repair_mod.repair_series(venue, repo, "BTC-USD", Granularity.ONE_DAY, now_ts=999)
+
+    assert len(result.errors) == 1
+    assert result.windows_absent_at_source == 0
+    assert repo.get_gap_probes("BTC-USD") == []
+
+    # The later, separate gap still got probed and filled in the same pass.
+    stored_ts = {c.ts for c in repo.get_candles("BTC-USD", Granularity.ONE_DAY)}
+    assert _BASE + 700 * _DAY in stored_ts
+    assert _BASE + 701 * _DAY in stored_ts
+
+    # Exactly the failed chunk's span -- not the whole original window -- remains missing.
+    (remaining,) = result.remaining
+    assert remaining.start_ts == _BASE + 300 * _DAY
+    assert remaining.end_ts == _BASE + 553 * _DAY
+    assert remaining.n_missing == 254
+
+
+def test_a_small_hole_still_takes_exactly_one_request(repo):
+    """Regression guard: chunking must not fragment requests that already fit under the cap."""
+    _seed(repo, missing={4, 5})
+    venue = _CappedVenue({_BASE + i * _DAY for i in range(10)})
+
+    result = repair_mod.repair_series(venue, repo, "BTC-USD", Granularity.ONE_DAY, now_ts=999)
+
+    assert len(venue.calls) == 1
+    assert result.bars_recovered == 2
+
+
+def test_sleep_fn_is_called_once_per_chunk_not_once_per_window(repo):
+    n_missing = 553
+    _seed(repo, missing=set(range(1, n_missing + 1)), n=n_missing + 2)
+    venue = _CappedVenue({_BASE + i * _DAY for i in range(n_missing + 2)})
+    sleeps: list[float] = []
+
+    repair_mod.repair_series(
+        venue, repo, "BTC-USD", Granularity.ONE_DAY, now_ts=999, sleep_fn=sleeps.append
+    )
+
+    assert len(sleeps) == 2
+
+
+def test_a_multi_chunk_window_with_nothing_at_venue_is_still_recorded_absent(repo):
+    """`probed_ok` requires every chunk to COMPLETE, not to return data. A large hole the venue
+    genuinely has none of must still end up recorded absent at source, exactly like a
+    single-chunk one would -- chunking is a fetch-mechanics detail, not a change in what counts
+    as a real probe."""
+    n_missing = 553
+    _seed(repo, missing=set(range(1, n_missing + 1)), n=n_missing + 2)
+    venue = _CappedVenue(set())  # the venue has none of the missing timestamps
+
+    result = repair_mod.repair_series(venue, repo, "BTC-USD", Granularity.ONE_DAY, now_ts=999)
+
+    assert result.windows_absent_at_source == 1
+    assert len(repo.get_gap_probes("BTC-USD")) == 1
+    assert len(venue.calls) > 1

tests/data/test_history.py

@@ -12,10 +12,13 @@
 
 import pytest
 
+from keel.data import history as history_mod
 from keel.data.db import connect, migrate
 from keel.data.history import (
     GRANULARITY_SECONDS,
     MAX_CANDLES_PER_REQUEST,
+    _fill_backward,
+    _fill_forward,
     coverage,
     ensure_history,
 )
@@ -110,3 +113,69 @@ def test_coverage_empty_when_nothing_cached(repo):
 
 def test_max_candles_per_request_is_conservative_under_coinbase_cap():
     assert MAX_CANDLES_PER_REQUEST == 300
+
+
+# -- inclusive-range off-by-one: [a, a + N*step] holds N+1 candles, not N ------
+#
+# Harmless at MAX_CANDLES_PER_REQUEST=300 (301 candles still clears Coinbase's ~350 ceiling),
+# but it silently activates the same incident as the repair-side hole in `repair.py` the moment
+# the constant is raised toward 350. Each window must hold at most MAX_CANDLES_PER_REQUEST
+# candles: `(end - start) // step + 1 <= MAX_CANDLES_PER_REQUEST`.
+
+
+def test_fill_forward_never_requests_more_than_the_cap_per_call(repo):
+    step = GRANULARITY_SECONDS[Granularity.ONE_HOUR]
+    latest_cached = 0
+    now = 1000 * step
+    full = [_mk(i * step) for i in range(1, 1001)]  # 1000 candles strictly newer than cached
+    client = FakeClient({"BTC-USD": full})
+
+    _fill_forward(
+        client, repo, "BTC-USD", Granularity.ONE_HOUR, step, latest_cached, now,
+        sleep_fn=lambda s: None, sleep_sec=0,
+    )
+
+    sizes = [(end - start) // step + 1 for (_, _, start, end) in client.calls]
+    assert sizes == [300, 300, 300, 100]
+    assert max(sizes) <= MAX_CANDLES_PER_REQUEST
+
+
+def test_fill_backward_never_requests_more_than_the_cap_per_call(repo):
+    step = GRANULARITY_SECONDS[Granularity.ONE_HOUR]
+    now = 2000 * step
+    window_end = now
+    start_floor = now - 999 * step
+    full = [_mk(now - i * step) for i in range(1000)]  # exactly [start_floor, window_end]
+    client = FakeClient({"BTC-USD": full})
+
+    _fill_backward(
+        client, repo, "BTC-USD", Granularity.ONE_HOUR, step, window_end, start_floor,
+        sleep_fn=lambda s: None, sleep_sec=0,
+    )
+
+    sizes = [(end - start) // step + 1 for (_, _, start, end) in client.calls]
+    assert sizes == [300, 300, 300, 100]
+    assert max(sizes) <= MAX_CANDLES_PER_REQUEST
+
+
+def test_window_sizing_tracks_the_cap_constant_so_raising_it_toward_350_stays_safe(
+    monkeypatch, repo
+):
+    """The whole point of the -1 fix: window sizing must be `(MAX_CANDLES_PER_REQUEST - 1) *
+    step`, derived from the constant, not a number that happens to match it today. Raise the cap
+    toward Coinbase's real ~350 ceiling and the per-call size must track it exactly, not drift
+    a candle over."""
+    monkeypatch.setattr(history_mod, "MAX_CANDLES_PER_REQUEST", 350)
+    step = GRANULARITY_SECONDS[Granularity.ONE_HOUR]
+    latest_cached = 0
+    now = 1000 * step
+    full = [_mk(i * step) for i in range(1, 1001)]
+    client = FakeClient({"BTC-USD": full})
+
+    history_mod._fill_forward(
+        client, repo, "BTC-USD", Granularity.ONE_HOUR, step, latest_cached, now,
+        sleep_fn=lambda s: None, sleep_sec=0,
+    )
+
+    sizes = [(end - start) // step + 1 for (_, _, start, end) in client.calls]
+    assert max(sizes) <= 350