From a31a86fe4bda57525f092d8fc3b22d22bf16f2ef Mon Sep 17 00:00:00 2001 From: Elmehdi Aitbrahim Date: Sun, 16 Aug 2026 17:30:37 -0400 Subject: [PATCH] fix(data): chunk `backfill`'s candle requests too, the last unpaged window site #269 chunked `market_feed.poll_once`; #271 swept for the same defect and fixed `repair.py` and `history.py`. `market_feed.backfill` was not reached by either, and has the identical shape: `_missing_ranges` groups absent timestamps into CONTIGUOUS ranges and each range was requested in a single call, so a hole wider than the venue's ~350-candle cap 400s exactly as the poll path did. On an empty repo the entire history window is one such range. LATENT, and stated plainly rather than dressed up: `backfill` has no production caller today -- `keel fetch` goes through `history.ensure_history` -- so nothing is failing on this right now. It is worth closing anyway on #271's own stated grounds, that every candle-request windowing site in the codebase should agree. A public data-layer entry point that 400s the moment it is called from anywhere is a trap left armed for whoever calls it next. The windowing arithmetic is now in one place, `_request_windows`, used by both `backfill` and `_poll_catch_up`. That was the actual reason this site was easy to miss: the `(MAX_CANDLES_PER_REQUEST - 1)` expression existed only inside `_poll_catch_up`, so `backfill` had nothing to be inconsistent WITH. #271 had to correct exactly that off-by-one after finding it duplicated in `history.py` -- the same bug class the shared `MAX_CANDLES_PER_REQUEST` import exists to prevent -- so it now has one definition here. `_poll_catch_up` keeps its own filter and `seen` dedup and is otherwise unchanged; folding it onto the helper also removed a `window_start = window_end + gran_sec` line left stranded when its `while` became a `for`. Tests written first and confirmed failing for the right reason: both cap tests failed on a single 575-candle request against the unfixed code. The third is a regression guard that a window within the cap still costs exactly one request, which passes before and after. Gates: ruff clean, mypy clean (77 source files), pytest 2762 -> 2765 passed / 1 skipped. Co-authored-by: Claude Opus 5 (1M context) --- keel/data/market_feed.py | 55 ++++++++++++++++++---------- tests/data/test_market_feed.py | 66 ++++++++++++++++++++++++++++++++++ 2 files changed, 102 insertions(+), 19 deletions(-) diff --git a/keel/data/market_feed.py b/keel/data/market_feed.py index c55eb345..096e66af 100644 --- a/keel/data/market_feed.py +++ b/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 diff --git a/tests/data/test_market_feed.py b/tests/data/test_market_feed.py index 5ed6d1d7..78ff0dcd 100644 --- a/tests/data/test_market_feed.py +++ b/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