diff --git a/keel/data/market_feed.py b/keel/data/market_feed.py index c55eb34..096e66a 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 5ed6d1d..78ff0dc 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