From c26c8d38487a42e73c320050c65a2b50da75bf5b Mon Sep 17 00:00:00 2001 From: Elmehdi Aitbrahim Date: Sat, 15 Aug 2026 16:04:13 -0400 Subject: [PATCH] fix(data): chunk the remaining candle-request windows under the venue cap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Coinbase rejects any candle request spanning more than ~350 candles with `400 INVALID_ARGUMENT ... "number of candles requested should be less than 350"`. #269 fixed this in `market_feed.poll_once`; review found two more sites with the same defect. Both are fixed here so every candle-request windowing site in the codebase agrees. Both are LATENT -- neither is firing today -- and both are the kind that fail quietly once they do. **`repair.py` cannot heal an interior hole larger than the cap.** `gaps.detect` puts no cap on `n_missing`, and `repair_series` asked for the whole hole in one call, so a gap of roughly 349+ candles 400s. Because the per-window `except` swallows the exception into `result.errors` and moves on, the failure would be *silent and permanent*: every scheduled repair retries the same oversized request, and the only symptom is a series that quietly never repairs. No cached series triggers this today. Measured across the deployment (24 products, ONE_HOUR and ONE_DAY), the largest interior gap is 15 bars (ICP-USD, ONE_HOUR) against a ~349 limit, and no series has an interior gap at or over the cap. So this is closing a latent trap, not healing an existing hole. It is still worth doing: an interior gap that large is entirely reachable -- a multi-day venue outage, a delisting and relisting, or a product added back after a long absence -- and the silent-forever failure mode is exactly the kind that is not noticed until much later. (For the avoidance of doubt: the 553-candle ZEC-USD figure from the #269 incident was end-of-series STALENESS, which `poll_once`/`ensure_history` catch up forward. It was never an interior gap, so `repair` was never asked to heal it. 553 survives here only as the oversized-gap size in the tests.) The widened window is now paged in `MAX_CANDLES_PER_REQUEST`-sized chunks (imported from `keel.data.history` -- a duplicated venue limit would be the same bug class), each upserted as it arrives so a mid-window failure leaves the earlier chunks persisted and the next run resumes further along. Semantics preserved exactly, because they are load-bearing: - the ±step widening applies to the OUTER edges only (venues disagree about endpoint inclusivity); internal chunk boundaries stay contiguous and are not themselves widened; - a window is eligible to be recorded *absent at source* only if EVERY chunk completed -- the difference between "the venue lacks this data" and "we failed to ask properly", and the module docstring is explicit that the absence record is an assertion about an observation; - the per-window `except ... continue` stays, so one bad window cannot abort the pass; - `sleep_fn` paces every chunk request, matching `history._fill_forward`. **`history.py`'s off-by-one would activate if the constant were raised.** An inclusive range `[a, a + N*step]` holds N+1 candles, so `_fill_forward` and `_fill_backward` requested 301, not 300 -- verified: a 1000-candle gap produced window sizes [301, 301, 301, 97]. Harmless at 300, but `history.py` invites raising the constant toward 350, at which point it would silently request 351 and reproduce the incident in the backfill path. Both sites now use `(MAX_CANDLES_PER_REQUEST - 1) * step`, matching what #269 did in `market_feed.py`, with a comment on the arithmetic so the -1 never reads as an accident. Tests written first and confirmed failing for the right reason: 9 of the 10 new tests fail against the unfixed code on candle-count, call-count and absence-record assertions (the 10th is a regression guard that a small hole still takes exactly one request, which must pass before and after). The repair tests drive an oversized gap against a fake venue that enforces the true ~349 ceiling, and assert on the actual start/end args it received. The history tests assert exact window sizes and express the invariant in terms of the constant, so raising it toward 350 stays safe. Gates: ruff clean, mypy clean (224 source files), pytest 2737 passed / 1 skipped (2727 + 1 skipped at 29d9b20, plus the 10 added here). Co-Authored-By: Claude Opus 5 (1M context) --- keel/data/history.py | 7 +- keel/data/repair.py | 62 ++++++++++++-- tests/data/test_gap_repair.py | 156 ++++++++++++++++++++++++++++++++++ tests/data/test_history.py | 69 +++++++++++++++ 4 files changed, 283 insertions(+), 11 deletions(-) diff --git a/keel/data/history.py b/keel/data/history.py index 4fe5e307..8493eb61 100644 --- a/keel/data/history.py +++ b/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 diff --git a/keel/data/repair.py b/keel/data/repair.py index 6fc3cf06..5f4389fd 100644 --- a/keel/data/repair.py +++ b/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)) diff --git a/tests/data/test_gap_repair.py b/tests/data/test_gap_repair.py index 633f7124..afd24e98 100644 --- a/tests/data/test_gap_repair.py +++ b/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 diff --git a/tests/data/test_history.py b/tests/data/test_history.py index a48dfa3b..f93d11c1 100644 --- a/tests/data/test_history.py +++ b/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