diff --git a/keel/data/market_feed.py b/keel/data/market_feed.py index c6476083..c55eb345 100644 --- a/keel/data/market_feed.py +++ b/keel/data/market_feed.py @@ -7,6 +7,13 @@ Only *closed* candles are ever fetched or persisted: the candle currently forming as of `now_ts` has OHLC values that are still changing, so both `backfill` and `poll_once` stop at the most recently closed candle boundary for each granularity. + +`poll_once`'s catch-up range is requested in windows of at most `MAX_CANDLES_PER_REQUEST` +candles, not as one single request: Coinbase rejects any range over ~350 candles with +`400 INVALID_ARGUMENT`. Without chunking, a product that ever falls behind that cap (e.g. after +downtime) would fail every poll forever -- the request only gets bigger over time, never +smaller -- so it can never self-heal. Chunking makes catch-up always succeed eventually, +however far behind a product has fallen. """ from __future__ import annotations @@ -14,6 +21,7 @@ import time from typing import TYPE_CHECKING +from keel.data.history import MAX_CANDLES_PER_REQUEST from keel.types import Candle, Granularity if TYPE_CHECKING: @@ -116,6 +124,46 @@ def backfill( return total_written +def _poll_catch_up( + client: CoinbaseClient, + repo: Repository, + product_id: str, + granularity: Granularity, + gran_sec: int, + fetch_start: int, + latest_closed: int, + last_ts: int | None, +) -> int: + """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. + """ + 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 + ) + fetched = client.get_candles(product_id, granularity, window_start, window_end) + new_candles: list[Candle] = [ + c + for c in fetched + if c.ts <= latest_closed and (last_ts is None or c.ts > last_ts) and c.ts not in seen + ] + 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 + + def poll_once( client: CoinbaseClient, repo: Repository, @@ -127,7 +175,11 @@ def poll_once( """Fetch and upsert any newly closed candles since the last poll. For each `(product, granularity)` pair, only candles strictly newer than the latest one - already stored (and no later than the most recently closed candle) are written. + already stored (and no later than the most recently closed candle) are written. The + catch-up range is requested in windows of at most `MAX_CANDLES_PER_REQUEST` candles (see + module docstring) so a product that has fallen far behind -- even hundreds of candles, as + happened in production -- still fully catches up in one call to `poll_once`, instead of + failing outright and staying stuck. Returns the total number of new candle rows written. """ @@ -145,14 +197,9 @@ def poll_once( continue fetch_start = last_ts + gran_sec if last_ts is not None else latest_closed - fetched = client.get_candles(product_id, granularity, fetch_start, latest_closed) - new_candles: list[Candle] = [ - c - for c in fetched - if c.ts <= latest_closed and (last_ts is None or c.ts > last_ts) - ] - if new_candles: - total_written += repo.upsert_candles(product_id, granularity, new_candles) + total_written += _poll_catch_up( + client, repo, product_id, granularity, gran_sec, fetch_start, latest_closed, last_ts + ) return total_written diff --git a/tests/data/test_market_feed.py b/tests/data/test_market_feed.py index 12f69e56..5ed6d1d7 100644 --- a/tests/data/test_market_feed.py +++ b/tests/data/test_market_feed.py @@ -12,6 +12,7 @@ import pytest from keel.data.db import connect, migrate +from keel.data.history import MAX_CANDLES_PER_REQUEST from keel.data.market_feed import backfill, is_fresh, poll_once from keel.data.repository import Repository from keel.types import Candle, Granularity @@ -23,6 +24,13 @@ WINDOW_START = NOW - HISTORY_DAYS * 86400 # already hour-aligned EXPECTED_TS = list(range(WINDOW_START, LATEST_CLOSED + 1, GRAN_SEC)) # 48 hourly candles +# A stale-poll scenario mirroring the real ZEC-USD production failure: the last stored +# candle is STALE_HOURS behind the most recently closed one, well over Coinbase's +# ~350-candle-per-request cap, so a correct `poll_once` must page the catch-up in windows. +STALE_HOURS = 552 # the real ZEC-USD gap +STALE_LAST_TS = LATEST_CLOSED - STALE_HOURS * GRAN_SEC # the stale last-stored candle +STALE_FULL_TS = list(range(STALE_LAST_TS, LATEST_CLOSED + 1, GRAN_SEC)) # seed + all catch-up + def _candle(ts: int, price: str = "100") -> Candle: p = Decimal(price) @@ -64,6 +72,13 @@ def _full_series(product_id: str = "BTC-USD") -> dict[tuple[str, Granularity], l return {(product_id, Granularity.ONE_HOUR): [_candle(ts) for ts in ts_values]} +def _stale_series(product_id: str = "BTC-USD") -> dict[tuple[str, Granularity], list[Candle]]: + # Contiguous hourly run from the stale last-stored candle through LATEST_CLOSED, plus + # the still-forming candle at ts=NOW, matching `_full_series`'s convention. + ts_values = STALE_FULL_TS + [NOW] + return {(product_id, Granularity.ONE_HOUR): [_candle(ts) for ts in ts_values]} + + # -- backfill ----------------------------------------------------------------- @@ -180,6 +195,104 @@ def test_poll_once_starts_from_scratch_when_repo_is_empty(repo): assert [c.ts for c in stored] == [LATEST_CLOSED] +def test_poll_once_uses_a_single_request_for_a_small_gap(repo): + # repo already has everything except the last 2 closed candles -- unchanged small-gap path + have = EXPECTED_TS[:-2] + repo.upsert_candles("BTC-USD", Granularity.ONE_HOUR, [_candle(ts) for ts in have]) + client = FakeClient(_full_series()) + + poll_once(client, repo, ["BTC-USD"], [Granularity.ONE_HOUR], now_ts=NOW) + + assert len(client.calls) == 1 + + +def test_poll_once_chunks_a_gap_larger_than_the_coinbase_cap(repo): + """A stale last candle (552h behind, like real ZEC-USD) must still be fully caught up.""" + repo.upsert_candles("BTC-USD", Granularity.ONE_HOUR, [_candle(STALE_LAST_TS)]) + client = FakeClient(_stale_series()) + + written = poll_once(client, repo, ["BTC-USD"], [Granularity.ONE_HOUR], now_ts=NOW) + + assert written == STALE_HOURS + assert len(client.calls) > 1 + stored = repo.get_candles("BTC-USD", Granularity.ONE_HOUR) + assert [c.ts for c in stored] == STALE_FULL_TS + stored_ts = {c.ts for c in stored} + assert NOW not in stored_ts # still-forming candle never persisted + + +def test_poll_once_never_requests_more_than_the_candle_cap(repo): + repo.upsert_candles("BTC-USD", Granularity.ONE_HOUR, [_candle(STALE_LAST_TS)]) + client = FakeClient(_stale_series()) + + poll_once(client, repo, ["BTC-USD"], [Granularity.ONE_HOUR], 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_poll_once_chunk_windows_are_contiguous_and_non_overlapping(repo): + repo.upsert_candles("BTC-USD", Granularity.ONE_HOUR, [_candle(STALE_LAST_TS)]) + client = FakeClient(_stale_series()) + + poll_once(client, repo, ["BTC-USD"], [Granularity.ONE_HOUR], now_ts=NOW) + + # a gap this large can only tile into >1 window under the Coinbase candle cap + assert len(client.calls) > 1 + fetch_start = STALE_LAST_TS + GRAN_SEC + assert client.calls[0][2] == fetch_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 + + +def test_poll_once_keeps_going_past_an_empty_window_so_a_mid_history_hole_cannot_wedge_catch_up( + repo, +): + """A window that returns zero candles must not stop catch-up. + + `_poll_catch_up`'s docstring calls this out explicitly: an empty window means a + mid-history hole (a stretch the venue genuinely has no candles for), not proof that + there's nothing further to fetch. If the loop broke on an empty window -- the way + `history._fill_backward` does -- catch-up would wedge at the hole forever and never + reach fresh data beyond it. That asymmetry with `_fill_backward` is intentional, not an + oversight: `_fill_backward` walks backward through history it may legitimately exhaust, + so stopping at "no more data" is correct there. `_poll_catch_up` walks forward toward + *now*, where there is always more recent data past any hole, so it must keep paging. + + The fixture spans a gap wider than the Coinbase cap (`STALE_HOURS`, as in the chunking + tests above), so catch-up must issue more than one windowed request. The fake client is + configured to have real candles for *none* of the first window's range -- an empty + window, simulating the hole -- and only serves candles for a later window. A future + refactor that adds `if not fetched: break` to `_poll_catch_up` makes this test fail: the + loop would stop after the first (empty) window and never reach the later, real candles. + """ + repo.upsert_candles("BTC-USD", Granularity.ONE_HOUR, [_candle(STALE_LAST_TS)]) + fetch_start = STALE_LAST_TS + GRAN_SEC + first_window_end = min( + LATEST_CLOSED, fetch_start + (MAX_CANDLES_PER_REQUEST - 1) * GRAN_SEC + ) + assert first_window_end < LATEST_CLOSED, "fixture must span >1 window for this test to hold" + later_ts = list(range(first_window_end + GRAN_SEC, LATEST_CLOSED + 1, GRAN_SEC)) + client = FakeClient( + {("BTC-USD", Granularity.ONE_HOUR): [_candle(ts) for ts in later_ts + [NOW]]} + ) + + written = poll_once(client, repo, ["BTC-USD"], [Granularity.ONE_HOUR], now_ts=NOW) + + # the empty first window must not have stopped the loop before a second request + assert len(client.calls) > 1 + assert written == len(later_ts) + stored_ts = {c.ts for c in repo.get_candles("BTC-USD", Granularity.ONE_HOUR)} + # the pre-existing seed candle plus exactly the later window's candles -- nothing from + # the empty (hole) window, since it had nothing to write + assert stored_ts == {STALE_LAST_TS, *later_ts} + assert LATEST_CLOSED in stored_ts # catch-up reached fresh data past the hole + + # -- is_fresh -------------------------------------------------------------------