diff --git a/src/fxhnt/application/bybit_concurrency.py b/src/fxhnt/application/bybit_concurrency.py index 2144500..b654356 100644 --- a/src/fxhnt/application/bybit_concurrency.py +++ b/src/fxhnt/application/bybit_concurrency.py @@ -38,6 +38,19 @@ BYBIT_RATE_LIMIT_RETCODES = frozenset({10006, 10018}) DEFAULT_WORKERS = 12 +def resume_floor_ms(last_day: dict[str, int], symbol: str, default_start_ms: int | None) -> int | None: + """Per-symbol backward-pagination floor for a DATE-RESUMING ingest: resume from the symbol's last persisted + day (`last_day[symbol] * 86_400_000`) so the adapter pulls only NEW days (the idempotent upsert overwrites + the boundary day in place). A symbol with no persisted day yet falls back to `default_start_ms` (the + caller's explicit floor, or None → the adapter's own backstop → full back-fill). + + This is the shared primitive behind the resume-from-last-date contract: it REPLACES the old symbol-skip + resume that, once a symbol carried ANY row, never fetched it again — which froze the feature at the last + ingest date and stalled the forward gate (see `last_day_by_symbol` on the feature store).""" + d = last_day.get(symbol) + return d * 86_400_000 if d is not None else default_start_ms + + class RateLimitError(RuntimeError): """Raised when a fetch stays rate-limited after exhausting the backoff retries.""" diff --git a/src/fxhnt/application/bybit_funding_ingest.py b/src/fxhnt/application/bybit_funding_ingest.py index 1727178..a94b224 100644 --- a/src/fxhnt/application/bybit_funding_ingest.py +++ b/src/fxhnt/application/bybit_funding_ingest.py @@ -5,13 +5,20 @@ feature='funding' at ts = epoch_day*86400 via the store's batched, idempotent up (one symbol at a time) and resumable-friendly (the upsert overwrites in place, so re-running is safe and cheap). Survivorship-free is the ADAPTER's job — a delisted symbol that returns rows is ingested like any other; one that returns nothing simply contributes no rows and never aborts the batch. + +DATE-RESUME (resume=True): each symbol is fetched FROM ITS LAST PERSISTED FUNDING DAY (not symbol-skipped), +so the nightly refresh advances funding every run instead of freezing it at the last ingest. """ from __future__ import annotations from collections.abc import Callable from typing import Any, Protocol -from fxhnt.application.bybit_concurrency import DEFAULT_WORKERS, fetch_symbols_concurrently +from fxhnt.application.bybit_concurrency import ( + DEFAULT_WORKERS, + fetch_symbols_concurrently, + resume_floor_ms, +) class _BybitFunding(Protocol): @@ -39,24 +46,22 @@ def ingest_bybit_funding( collected up front, so peak memory stays ~one symbol (preserving the OOM-fix property). The fetch is the bottleneck, so concurrent-fetch + serialized-write gives ~5–10× over the old sequential loop. - Resumable (`resume=True`, default): symbols already carrying the 'funding' feature in `store` are SKIPPED - — the destination is queried ONCE up front (before the pool) for the done-set, so a re-run after a partial - ingest (e.g. an OOM kill at symbol N) fetches ONLY the remaining symbols. The summary counts only this - run's writes; the overlap is left untouched. The skip is purely a fetch-cost optimisation: because the - upsert is idempotent, `resume=False` re-fetches and re-writes everything to the identical final state.""" + Resumable (`resume=True`, default): each symbol resumes FROM ITS LAST PERSISTED 'funding' DAY — the + per-symbol last day is queried ONCE up front (`store.last_day_by_symbol('funding')`) and used as the + backward-pagination floor, so a symbol already carrying funding is re-fetched only from its last day + forward (advancing it with new days; the idempotent upsert overwrites the boundary day in place) and a + fresh symbol is back-filled from `start_ms` (or the adapter's backstop). NOTHING is symbol-skipped, so + funding advances every run. `resume=False` ignores the persisted days and re-fetches everything from + `start_ms` to the identical final state (idempotent).""" n_symbols = 0 day_rows = 0 oldest_day: int | None = None - todo = symbols - if resume: - done = store.symbols_with_feature("funding") - if done: - todo = [s for s in symbols if s not in done] - if progress is not None: - progress(f"resuming: {len(symbols) - len(todo)} already done, {len(todo)} to go") + last_day = store.last_day_by_symbol("funding") if resume else {} + if resume and last_day and progress is not None: + progress(f"resuming from last persisted date: {len(last_day)} symbols already carry funding") def fetch_one(sym: str) -> dict[int, float]: - return bybit.funding_history(sym, start_ms=start_ms) + return bybit.funding_history(sym, start_ms=resume_floor_ms(last_day, sym, start_ms)) def on_result(sym: str, daily: dict[int, float]) -> None: nonlocal n_symbols, day_rows, oldest_day @@ -73,5 +78,5 @@ def ingest_bybit_funding( if progress is not None: progress(f"{sym}: {len(rows)} day-rows (oldest day {sym_oldest})") - fetch_symbols_concurrently(todo, fetch_one, workers=workers, on_result=on_result) + fetch_symbols_concurrently(symbols, fetch_one, workers=workers, on_result=on_result) return {"symbols": n_symbols, "day_rows": day_rows, "oldest_day": oldest_day} diff --git a/src/fxhnt/application/bybit_klines_ingest.py b/src/fxhnt/application/bybit_klines_ingest.py index d8c63e2..77d924f 100644 --- a/src/fxhnt/application/bybit_klines_ingest.py +++ b/src/fxhnt/application/bybit_klines_ingest.py @@ -23,7 +23,11 @@ from __future__ import annotations from collections.abc import Callable from typing import Any, Protocol -from fxhnt.application.bybit_concurrency import DEFAULT_WORKERS, fetch_symbols_concurrently +from fxhnt.application.bybit_concurrency import ( + DEFAULT_WORKERS, + fetch_symbols_concurrently, + resume_floor_ms, +) class _BybitKlines(Protocol): @@ -148,9 +152,8 @@ def ingest_bybit_spot( newest_day: int | None = None def fetch_one(sym: str) -> dict[int, dict[str, float]]: - d = last_day.get(sym) - start_ms = d * 86_400_000 if d is not None else None # resume from the last persisted spot day - return bybit.daily_ohlcv(sym, category="spot", start_ms=start_ms) + # Resume from the last persisted spot day (shared primitive); fresh symbols back-fill from the floor. + return bybit.daily_ohlcv(sym, category="spot", start_ms=resume_floor_ms(last_day, sym, None)) def on_result(sym: str, spot: dict[int, dict[str, float]]) -> None: nonlocal n_symbols, spot_rows, newest_day diff --git a/src/fxhnt/application/bybit_long_short_ingest.py b/src/fxhnt/application/bybit_long_short_ingest.py index 86e5227..694d265 100644 --- a/src/fxhnt/application/bybit_long_short_ingest.py +++ b/src/fxhnt/application/bybit_long_short_ingest.py @@ -2,9 +2,10 @@ Per symbol: fetch {epoch_day: buy_ratio} (the adapter; period=1d → one value/day), write feature 'long_ratio' at ts = epoch_day*86400 via the store's batched, idempotent upsert. Memory-bounded (results are -written as completed, never collected up front — ~one symbol's payload peak) and resumable (the done-set = -symbols already carrying 'long_ratio'). ONLY the long_ratio feature is written — funding/close/spot are the -job of the other Bybit ingests; this one never touches them. +written as completed, never collected up front — ~one symbol's payload peak) and DATE-RESUMABLE (each symbol +resumes from its last persisted 'long_ratio' day, so the feature advances every run instead of freezing). ONLY +the long_ratio feature is written — funding/close/spot are the job of the other Bybit ingests; this one never +touches them. Mirrors `bybit_funding_ingest.ingest_bybit_funding` (concurrent FETCH across a bounded pool + serialized main-thread WRITE) — same OOM-safe, resume-on-partial contract. @@ -14,7 +15,11 @@ from __future__ import annotations from collections.abc import Callable from typing import Any, Protocol -from fxhnt.application.bybit_concurrency import DEFAULT_WORKERS, fetch_symbols_concurrently +from fxhnt.application.bybit_concurrency import ( + DEFAULT_WORKERS, + fetch_symbols_concurrently, + resume_floor_ms, +) class _BybitAccountRatio(Protocol): @@ -42,23 +47,21 @@ def ingest_bybit_long_short( (serialized — no store thread-safety needed). Memory-bounded: results are written as completed, never collected up front, so peak memory stays ~one symbol. - Resumable (`resume=True`, default): symbols already carrying the 'long_ratio' feature in `store` are - SKIPPED — the destination is queried ONCE up front (before the pool) for the done-set, so a re-run after a - partial ingest fetches ONLY the remaining symbols. The summary counts only this run's writes. Because the - upsert is idempotent, `resume=False` re-fetches and re-writes everything to the identical final state.""" + Resumable (`resume=True`, default): each symbol resumes FROM ITS LAST PERSISTED 'long_ratio' DAY — the + per-symbol last day is queried ONCE up front (`store.last_day_by_symbol('long_ratio')`) and used as the + backward-pagination floor, so a symbol already carrying long_ratio is re-fetched only from its last day + forward (advancing it; the idempotent upsert overwrites the boundary day in place) and a fresh symbol is + back-filled from `start_ms`. NOTHING is symbol-skipped, so long_ratio advances every run. Because the + upsert is idempotent, `resume=False` re-fetches everything from `start_ms` to the identical final state.""" n_symbols = 0 day_rows = 0 oldest_day: int | None = None - todo = symbols - if resume: - done = store.symbols_with_feature("long_ratio") - if done: - todo = [s for s in symbols if s not in done] - if progress is not None: - progress(f"resuming: {len(symbols) - len(todo)} already done, {len(todo)} to go") + last_day = store.last_day_by_symbol("long_ratio") if resume else {} + if resume and last_day and progress is not None: + progress(f"resuming from last persisted date: {len(last_day)} symbols already carry long_ratio") def fetch_one(sym: str) -> dict[int, float]: - return bybit.long_ratio_history(sym, start_ms=start_ms) + return bybit.long_ratio_history(sym, start_ms=resume_floor_ms(last_day, sym, start_ms)) def on_result(sym: str, daily: dict[int, float]) -> None: nonlocal n_symbols, day_rows, oldest_day @@ -75,5 +78,5 @@ def ingest_bybit_long_short( if progress is not None: progress(f"{sym}: {len(rows)} day-rows (oldest day {sym_oldest})") - fetch_symbols_concurrently(todo, fetch_one, workers=workers, on_result=on_result) + fetch_symbols_concurrently(symbols, fetch_one, workers=workers, on_result=on_result) return {"symbols": n_symbols, "day_rows": day_rows, "oldest_day": oldest_day} diff --git a/src/fxhnt/application/bybit_oi_ingest.py b/src/fxhnt/application/bybit_oi_ingest.py index f6b038c..6da1989 100644 --- a/src/fxhnt/application/bybit_oi_ingest.py +++ b/src/fxhnt/application/bybit_oi_ingest.py @@ -2,9 +2,10 @@ Per symbol: fetch {epoch_day: open_interest} (the adapter; intervalTime=1d → one value/day), write feature 'open_interest' at ts = epoch_day*86400 via the store's batched, idempotent upsert. Memory-bounded (results -are written as completed, never collected up front — ~one symbol's payload peak) and resumable (the done-set = -symbols already carrying 'open_interest'). ONLY the open_interest feature is written — funding/close/spot/ -long_ratio are the job of the other Bybit ingests; this one never touches them. +are written as completed, never collected up front — ~one symbol's payload peak) and DATE-RESUMABLE (each +symbol resumes from its last persisted 'open_interest' day, so the feature advances every run instead of +freezing). ONLY the open_interest feature is written — funding/close/spot/long_ratio are the job of the other +Bybit ingests; this one never touches them. Mirrors `bybit_long_short_ingest.ingest_bybit_long_short` (concurrent FETCH across a bounded pool + serialized main-thread WRITE) — same OOM-safe, resume-on-partial contract. @@ -14,7 +15,11 @@ from __future__ import annotations from collections.abc import Callable from typing import Any, Protocol -from fxhnt.application.bybit_concurrency import DEFAULT_WORKERS, fetch_symbols_concurrently +from fxhnt.application.bybit_concurrency import ( + DEFAULT_WORKERS, + fetch_symbols_concurrently, + resume_floor_ms, +) class _BybitOpenInterest(Protocol): @@ -42,23 +47,21 @@ def ingest_bybit_oi( (serialized — no store thread-safety needed). Memory-bounded: results are written as completed, never collected up front, so peak memory stays ~one symbol. - Resumable (`resume=True`, default): symbols already carrying the 'open_interest' feature in `store` are - SKIPPED — the destination is queried ONCE up front (before the pool) for the done-set, so a re-run after a - partial ingest fetches ONLY the remaining symbols. The summary counts only this run's writes. Because the - upsert is idempotent, `resume=False` re-fetches and re-writes everything to the identical final state.""" + Resumable (`resume=True`, default): each symbol resumes FROM ITS LAST PERSISTED 'open_interest' DAY — the + per-symbol last day is queried ONCE up front (`store.last_day_by_symbol('open_interest')`) and used as the + backward-pagination floor, so a symbol already carrying open_interest is re-fetched only from its last day + forward (advancing it; the idempotent upsert overwrites the boundary day in place) and a fresh symbol is + back-filled from `start_ms`. NOTHING is symbol-skipped, so open_interest advances every run. Because the + upsert is idempotent, `resume=False` re-fetches everything from `start_ms` to the identical final state.""" n_symbols = 0 day_rows = 0 oldest_day: int | None = None - todo = symbols - if resume: - done = store.symbols_with_feature("open_interest") - if done: - todo = [s for s in symbols if s not in done] - if progress is not None: - progress(f"resuming: {len(symbols) - len(todo)} already done, {len(todo)} to go") + last_day = store.last_day_by_symbol("open_interest") if resume else {} + if resume and last_day and progress is not None: + progress(f"resuming from last persisted date: {len(last_day)} symbols already carry open_interest") def fetch_one(sym: str) -> dict[int, float]: - return bybit.oi_history(sym, start_ms=start_ms) + return bybit.oi_history(sym, start_ms=resume_floor_ms(last_day, sym, start_ms)) def on_result(sym: str, daily: dict[int, float]) -> None: nonlocal n_symbols, day_rows, oldest_day @@ -75,5 +78,5 @@ def ingest_bybit_oi( if progress is not None: progress(f"{sym}: {len(rows)} day-rows (oldest day {sym_oldest})") - fetch_symbols_concurrently(todo, fetch_one, workers=workers, on_result=on_result) + fetch_symbols_concurrently(symbols, fetch_one, workers=workers, on_result=on_result) return {"symbols": n_symbols, "day_rows": day_rows, "oldest_day": oldest_day} diff --git a/src/fxhnt/application/bybit_worst_basis_ingest.py b/src/fxhnt/application/bybit_worst_basis_ingest.py index e79a857..d51726f 100644 --- a/src/fxhnt/application/bybit_worst_basis_ingest.py +++ b/src/fxhnt/application/bybit_worst_basis_ingest.py @@ -13,15 +13,20 @@ Concurrency (`workers`, default 12): the per-symbol fetch (network-bound 1m pagi across the shared bounded pool (`fetch_symbols_concurrently`); each symbol's WRITE happens in the main thread as the fetch completes (serialized — no store thread-safety needed). MEMORY-BOUNDED: results are written as-completed, never collected up front (~one symbol's day-series resident); the adapter itself is per-day -bounded. RESUME (`resume=True`): symbols already carrying `worst_basis` are skipped (the done-criterion), so a -partial/OOM-killed run resumes cleanly. Idempotent upsert ⇒ `resume=False` re-fetches to the identical state. +bounded. DATE-RESUME (`resume=True`): each symbol resumes FROM ITS LAST PERSISTED `worst_basis` DAY (not +symbol-skipped), so the heavy weekly job advances every run instead of freezing at the last ingest. Idempotent +upsert ⇒ `resume=False` re-fetches to the identical state. """ from __future__ import annotations from collections.abc import Callable from typing import Any, Protocol -from fxhnt.application.bybit_concurrency import DEFAULT_WORKERS, fetch_symbols_concurrently +from fxhnt.application.bybit_concurrency import ( + DEFAULT_WORKERS, + fetch_symbols_concurrently, + resume_floor_ms, +) class _BybitKlines(Protocol): @@ -46,22 +51,22 @@ def ingest_bybit_worst_basis( day-entries written, W = the single deepest (most negative) worst_basis observed (0.0 if none). A symbol with no overlap days contributes nothing and never aborts. - Resumable: with `resume=True`, symbols already carrying the `worst_basis` feature in `store` are skipped - (the destination is queried ONCE up front, before the pool). The summary counts only this run's writes. + Resumable: with `resume=True`, each symbol resumes FROM ITS LAST PERSISTED `worst_basis` DAY — the + per-symbol last day is queried ONCE up front (`store.last_day_by_symbol('worst_basis')`) and used as the + forward-window floor, so a symbol already carrying worst_basis is re-computed only from its last day + forward (advancing it; idempotent upsert) and a fresh symbol is computed from `start_ms`. NOTHING is + symbol-skipped, so worst_basis advances every run. The summary counts only this run's writes. """ n_symbols = 0 total_days = 0 worst = 0.0 - todo = symbols - if resume: - done = store.symbols_with_feature("worst_basis") - if done: - todo = [s for s in symbols if s not in done] - if progress is not None: - progress(f"resuming: {len(symbols) - len(todo)} already done, {len(todo)} to go") + last_day = store.last_day_by_symbol("worst_basis") if resume else {} + if resume and last_day and progress is not None: + progress(f"resuming from last persisted date: {len(last_day)} symbols already carry worst_basis") def fetch_one(sym: str) -> dict[int, float]: - return bybit.intraday_worst_basis(sym, start_ms=start_ms, interval=interval) + return bybit.intraday_worst_basis( + sym, start_ms=resume_floor_ms(last_day, sym, start_ms), interval=interval) def on_result(sym: str, wb: dict[int, float]) -> None: nonlocal n_symbols, total_days, worst @@ -76,5 +81,5 @@ def ingest_bybit_worst_basis( if progress is not None: progress(f"{sym}: {len(wb)} worst_basis days (worst {min(wb.values()):.4f})") - fetch_symbols_concurrently(todo, fetch_one, workers=workers, on_result=on_result) + fetch_symbols_concurrently(symbols, fetch_one, workers=workers, on_result=on_result) return {"symbols": n_symbols, "days": total_days, "worst": worst} diff --git a/tests/integration/test_bybit_forward_track.py b/tests/integration/test_bybit_forward_track.py index ddc842b..9751992 100644 --- a/tests/integration/test_bybit_forward_track.py +++ b/tests/integration/test_bybit_forward_track.py @@ -218,13 +218,12 @@ class _FakeAccountRatio: return {19000: 0.5, 19001: 0.5} -def test_refresh_skips_already_done_symbols_in_resume_mode() -> None: - """A symbol already carrying a feature is NOT re-fetched for that leg (the symbol-skip resume contract of - the perp/funding/long_ratio ingests) — proven by the mock adapters never seeing the pre-seeded symbol's - PERP / funding fetch. The SPOT leg is the deliberate exception: it date-resumes (see the spot-advance - test), so a symbol lacking spot_close is back-filled even when its perp leg is done.""" +def test_refresh_perp_symbol_skips_while_other_legs_date_resume() -> None: + """Mixed resume contract after the date-resume migration: the PERP (linear) klines leg still SYMBOL-SKIPS + a symbol already carrying `turnover`, but every other per-symbol leg (spot/funding/long_ratio) DATE-RESUMES + — so a done symbol is still fetched for those legs (from its last persisted day) and advances every run.""" store = TimescaleFeatureStore("sqlite://", table="bybit_features") - # Pre-seed DONE (perp/funding/long_ratio) but NO spot_close → perp/funding skip, spot back-fills. + # Pre-seed DONE (perp turnover + funding + long_ratio) but NO spot_close. store.write_features("BTCUSDT", [(19000 * _DAY, {"close": 100.0, "turnover": 9e9, "funding": 0.001, "long_ratio": 0.5})]) @@ -235,13 +234,12 @@ def test_refresh_skips_already_done_symbols_in_resume_mode() -> None: universe=universe, min_dollar_vol=1_000_000.0) store.close() - # BTCUSDT perp leg was done (carries turnover) → its PERP (linear) fetch is skipped; ETHUSDT is fetched. + # BTCUSDT perp leg was done (carries turnover) → its PERP (linear) fetch is symbol-skipped; ETHUSDT fetched. assert ("BTCUSDT", "linear") not in klines.calls assert ("ETHUSDT", "linear") in klines.calls - # BTCUSDT lacks spot_close → the dedicated spot refresh DOES fetch its spot leg (back-fill). + # spot + funding DATE-RESUME → the done symbol IS fetched for those legs (advance, not skip). assert ("BTCUSDT", "spot") in klines.calls - # funding still symbol-skips the done symbol. - assert "BTCUSDT" not in funding.calls + assert "BTCUSDT" in funding.calls assert "ETHUSDT" in funding.calls assert summary["klines"]["symbols"] >= 1 diff --git a/tests/integration/test_bybit_funding_ingest.py b/tests/integration/test_bybit_funding_ingest.py index 68a4722..1ed7080 100644 --- a/tests/integration/test_bybit_funding_ingest.py +++ b/tests/integration/test_bybit_funding_ingest.py @@ -73,33 +73,36 @@ def test_ingest_progress_callback_invoked(): store.close() -def test_resume_skips_already_done_symbols_and_completes_the_rest(): - """Ingest {A,B}, then ingest {A,B,C,D} with resume=True -> ONLY C,D are fetched (the adapter is NOT - called for A,B), and the final store holds all four. This is the OOM-resume contract.""" +def test_resume_advances_done_symbols_from_last_date_and_completes_the_rest(): + """DATE-RESUME (replaces symbol-skip): A,B already carry OLD funding -> re-fetched FROM THEIR LAST + PERSISTED DAY so NEW days land (advance), while C,D are back-filled from the default floor. Nothing is + skipped, so funding advances every run (the freeze fix).""" store = TimescaleFeatureStore("sqlite://", table="bybit_features") + # First (partial) run seeds A,B at day 19000 only. + ingest_bybit_funding(store, _FakeBybit({"AUSDT": {19000: 0.1}, "BUSDT": {19000: 0.2}}), + ["AUSDT", "BUSDT"]) + + # Resume run: the adapter now serves an EXTRA day (19001) for A,B and full history for C,D. panels = { - "AUSDT": {19000: 0.1}, "BUSDT": {19000: 0.2}, + "AUSDT": {19000: 0.1, 19001: 0.15}, "BUSDT": {19000: 0.2, 19001: 0.25}, "CUSDT": {19000: 0.3}, "DUSDT": {19000: 0.4}, } - # First (partial) run: only A, B. - first = _FakeBybit(panels) - ingest_bybit_funding(store, first, ["AUSDT", "BUSDT"]) - assert {c[0] for c in first.calls} == {"AUSDT", "BUSDT"} - - # Resume run over the full universe: A, B are already done -> must NOT be fetched again. second = _FakeBybit(panels) seen: list[str] = [] - summary = ingest_bybit_funding( + ingest_bybit_funding( store, second, ["AUSDT", "BUSDT", "CUSDT", "DUSDT"], resume=True, progress=lambda m: seen.append(m)) - fetched = {c[0] for c in second.calls} - assert fetched == {"CUSDT", "DUSDT"}, f"resume re-fetched done symbols: {fetched}" - assert summary["symbols"] == 2 and summary["day_rows"] == 2 # only this run's writes - assert any("resuming: 2 already done, 2 to go" in m for m in seen) - # Final state is the union of all four. + # ALL four fetched (no skip); A,B resume from their last persisted day, C,D from the default floor (None). + fetched = dict(second.calls) + assert set(fetched) == {"AUSDT", "BUSDT", "CUSDT", "DUSDT"} + assert fetched["AUSDT"] == 19000 * 86_400_000 and fetched["BUSDT"] == 19000 * 86_400_000 + assert fetched["CUSDT"] is None and fetched["DUSDT"] is None + assert any("resuming from last persisted date" in m for m in seen) + + # The done symbols ADVANCED to the new day; the rest were back-filled. assert store.crypto_funding_panel() == { - "AUSDT": {19000: 0.1}, "BUSDT": {19000: 0.2}, + "AUSDT": {19000: 0.1, 19001: 0.15}, "BUSDT": {19000: 0.2, 19001: 0.25}, "CUSDT": {19000: 0.3}, "DUSDT": {19000: 0.4}, } store.close() @@ -151,15 +154,19 @@ def test_concurrent_ingest_writes_all_symbols_exactly_once(): store.close() -def test_concurrent_resume_skips_done_set_computed_before_pool(): - """Resume still works under concurrency: the done-set is queried once up front, so a workers>1 resume - re-fetches ONLY the remaining symbols.""" +def test_concurrent_resume_refetches_all_from_last_date(): + """Under concurrency, date-resume re-fetches ALL symbols (the done half from their last persisted day), + landing the full panel — no symbol is skipped (the per-symbol last-day map is queried once up front).""" store = TimescaleFeatureStore("sqlite://", table="bybit_features") panels = {f"S{i}USDT": {19000: float(i)} for i in range(20)} - ingest_bybit_funding(store, _FakeBybit(panels), list(panels)[:10], workers=8) + done_half = list(panels)[:10] + ingest_bybit_funding(store, _FakeBybit(panels), done_half, workers=8) second = _FakeBybit(panels) ingest_bybit_funding(store, second, list(panels), resume=True, workers=8) - assert {c[0] for c in second.calls} == set(list(panels)[10:]) # only the not-yet-done half + fetched = dict(second.calls) + assert set(fetched) == set(panels) # ALL fetched, none skipped + assert all(fetched[s] == 19000 * 86_400_000 for s in done_half) # done half resumes from last date + assert all(fetched[s] is None for s in list(panels)[10:]) # fresh half from the default floor assert store.crypto_funding_panel() == {s: d for s, d in panels.items()} store.close() diff --git a/tests/integration/test_bybit_long_short_ingest.py b/tests/integration/test_bybit_long_short_ingest.py index 0ab0d33..5c75120 100644 --- a/tests/integration/test_bybit_long_short_ingest.py +++ b/tests/integration/test_bybit_long_short_ingest.py @@ -56,25 +56,28 @@ def test_ingest_only_writes_long_ratio_feature_not_funding_or_close(): store.close() -def test_resume_skips_already_done_symbols_and_completes_the_rest(): +def test_resume_advances_done_symbols_from_last_date_and_completes_the_rest(): + """DATE-RESUME (replaces symbol-skip): A,B already carry OLD long_ratio -> re-fetched FROM THEIR LAST + PERSISTED DAY so NEW days land (advance); C,D back-filled from the default floor. Nothing is skipped.""" store = TimescaleFeatureStore("sqlite://", table="bybit_features") + ingest_bybit_long_short(store, _FakeBybit({"AUSDT": {19000: 0.5}, "BUSDT": {19000: 0.5}}), + ["AUSDT", "BUSDT"]) + panels = { - "AUSDT": {19000: 0.5}, "BUSDT": {19000: 0.5}, + "AUSDT": {19000: 0.5, 19001: 0.55}, "BUSDT": {19000: 0.5, 19001: 0.45}, "CUSDT": {19000: 0.5}, "DUSDT": {19000: 0.5}, } - first = _FakeBybit(panels) - ingest_bybit_long_short(store, first, ["AUSDT", "BUSDT"]) - assert {c[0] for c in first.calls} == {"AUSDT", "BUSDT"} - second = _FakeBybit(panels) seen: list[str] = [] - summary = ingest_bybit_long_short( + ingest_bybit_long_short( store, second, ["AUSDT", "BUSDT", "CUSDT", "DUSDT"], resume=True, progress=lambda m: seen.append(m)) - fetched = {c[0] for c in second.calls} - assert fetched == {"CUSDT", "DUSDT"}, f"resume re-fetched done symbols: {fetched}" - assert summary["symbols"] == 2 and summary["day_rows"] == 2 - assert any("resuming: 2 already done, 2 to go" in m for m in seen) + fetched = dict(second.calls) + assert set(fetched) == {"AUSDT", "BUSDT", "CUSDT", "DUSDT"} + assert fetched["AUSDT"] == 19000 * 86_400_000 and fetched["BUSDT"] == 19000 * 86_400_000 + assert fetched["CUSDT"] is None and fetched["DUSDT"] is None + assert any("resuming from last persisted date" in m for m in seen) + assert store.read_panel(["AUSDT", "BUSDT", "CUSDT", "DUSDT"], "long_ratio") == panels store.close() @@ -100,12 +103,18 @@ def test_concurrent_ingest_writes_all_symbols_exactly_once(): store.close() -def test_concurrent_resume_skips_done_set_computed_before_pool(): +def test_concurrent_resume_refetches_all_from_last_date(): + """Under concurrency, date-resume re-fetches ALL symbols (the done half from their last persisted day) — + no symbol is skipped (the per-symbol last-day map is queried once up front).""" store = TimescaleFeatureStore("sqlite://", table="bybit_features") panels = {f"S{i}USDT": {19000: 0.5} for i in range(20)} - ingest_bybit_long_short(store, _FakeBybit(panels), list(panels)[:10], workers=8) + done_half = list(panels)[:10] + ingest_bybit_long_short(store, _FakeBybit(panels), done_half, workers=8) second = _FakeBybit(panels) ingest_bybit_long_short(store, second, list(panels), resume=True, workers=8) - assert {c[0] for c in second.calls} == set(list(panels)[10:]) # only the not-yet-done half + fetched = dict(second.calls) + assert set(fetched) == set(panels) + assert all(fetched[s] == 19000 * 86_400_000 for s in done_half) + assert all(fetched[s] is None for s in list(panels)[10:]) store.close() diff --git a/tests/integration/test_bybit_oi_ingest.py b/tests/integration/test_bybit_oi_ingest.py index b7231db..6f8077c 100644 --- a/tests/integration/test_bybit_oi_ingest.py +++ b/tests/integration/test_bybit_oi_ingest.py @@ -57,25 +57,27 @@ def test_ingest_only_writes_open_interest_feature_not_funding_or_close(): store.close() -def test_resume_skips_already_done_symbols_and_completes_the_rest(): +def test_resume_advances_done_symbols_from_last_date_and_completes_the_rest(): + """DATE-RESUME (replaces symbol-skip): A,B already carry OLD open_interest -> re-fetched FROM THEIR LAST + PERSISTED DAY so NEW days land (advance); C,D back-filled from the default floor. Nothing is skipped.""" store = TimescaleFeatureStore("sqlite://", table="bybit_features") + ingest_bybit_oi(store, _FakeBybit({"AUSDT": {19000: 1.0}, "BUSDT": {19000: 1.0}}), ["AUSDT", "BUSDT"]) + panels = { - "AUSDT": {19000: 1.0}, "BUSDT": {19000: 1.0}, + "AUSDT": {19000: 1.0, 19001: 1.5}, "BUSDT": {19000: 1.0, 19001: 1.6}, "CUSDT": {19000: 1.0}, "DUSDT": {19000: 1.0}, } - first = _FakeBybit(panels) - ingest_bybit_oi(store, first, ["AUSDT", "BUSDT"]) - assert {c[0] for c in first.calls} == {"AUSDT", "BUSDT"} - second = _FakeBybit(panels) seen: list[str] = [] - summary = ingest_bybit_oi( + ingest_bybit_oi( store, second, ["AUSDT", "BUSDT", "CUSDT", "DUSDT"], resume=True, progress=lambda m: seen.append(m)) - fetched = {c[0] for c in second.calls} - assert fetched == {"CUSDT", "DUSDT"}, f"resume re-fetched done symbols: {fetched}" - assert summary["symbols"] == 2 and summary["day_rows"] == 2 - assert any("resuming: 2 already done, 2 to go" in m for m in seen) + fetched = dict(second.calls) + assert set(fetched) == {"AUSDT", "BUSDT", "CUSDT", "DUSDT"} + assert fetched["AUSDT"] == 19000 * 86_400_000 and fetched["BUSDT"] == 19000 * 86_400_000 + assert fetched["CUSDT"] is None and fetched["DUSDT"] is None + assert any("resuming from last persisted date" in m for m in seen) + assert store.read_panel(["AUSDT", "BUSDT", "CUSDT", "DUSDT"], "open_interest") == panels store.close() @@ -100,12 +102,18 @@ def test_concurrent_ingest_writes_all_symbols_exactly_once(): store.close() -def test_concurrent_resume_skips_done_set_computed_before_pool(): +def test_concurrent_resume_refetches_all_from_last_date(): + """Under concurrency, date-resume re-fetches ALL symbols (the done half from their last persisted day) — + no symbol is skipped (the per-symbol last-day map is queried once up front).""" store = TimescaleFeatureStore("sqlite://", table="bybit_features") panels = {f"S{i}USDT": {19000: 1.0} for i in range(20)} - ingest_bybit_oi(store, _FakeBybit(panels), list(panels)[:10], workers=8) + done_half = list(panels)[:10] + ingest_bybit_oi(store, _FakeBybit(panels), done_half, workers=8) second = _FakeBybit(panels) ingest_bybit_oi(store, second, list(panels), resume=True, workers=8) - assert {c[0] for c in second.calls} == set(list(panels)[10:]) # only the not-yet-done half + fetched = dict(second.calls) + assert set(fetched) == set(panels) + assert all(fetched[s] == 19000 * 86_400_000 for s in done_half) + assert all(fetched[s] is None for s in list(panels)[10:]) store.close() diff --git a/tests/integration/test_bybit_worst_basis_ingest.py b/tests/integration/test_bybit_worst_basis_ingest.py index 1610848..82839a8 100644 --- a/tests/integration/test_bybit_worst_basis_ingest.py +++ b/tests/integration/test_bybit_worst_basis_ingest.py @@ -1,8 +1,8 @@ """ingest_bybit_worst_basis — per symbol compute the daily 1m worst_basis series and write the `worst_basis` feature into a bybit_features TimescaleFeatureStore. -Resumable (done-set = symbols already carrying `worst_basis`), concurrent (all symbols fetched once via the -shared pool), bounded-memory (one symbol at a time). The Bybit adapter is a fake — NO HTTP. +Date-resumable (each symbol resumes from its last persisted `worst_basis` day), concurrent (all symbols +fetched once via the shared pool), bounded-memory (one symbol at a time). The Bybit adapter is a fake — NO HTTP. """ from __future__ import annotations @@ -17,10 +17,10 @@ class _FakeBybit: def __init__(self, panels: dict[str, dict[int, float]]) -> None: self._panels = panels - self.calls: list[str] = [] + self.calls: list[tuple[str, int | None]] = [] # (symbol, start_ms) — to assert date-resume def intraday_worst_basis(self, symbol: str, *, start_ms=None, interval: str = "1") -> dict[int, float]: - self.calls.append(symbol) + self.calls.append((symbol, start_ms)) return self._panels.get(symbol, {}) @@ -53,24 +53,28 @@ def test_ingest_is_idempotent() -> None: store.close() -def test_resume_skips_symbols_with_worst_basis() -> None: +def test_resume_advances_done_symbols_from_last_date_and_completes_the_rest() -> None: + """DATE-RESUME (replaces symbol-skip): A,B already carry OLD worst_basis -> re-computed FROM THEIR LAST + PERSISTED DAY so NEW days land (advance); C,D back-filled from the default floor. Nothing is skipped, so + the heavy weekly job advances every run.""" store = TimescaleFeatureStore("sqlite://", table="bybit_features") - panels = {"AUSDT": {19000: -0.1}, "BUSDT": {19000: -0.2}, - "CUSDT": {19000: -0.3}, "DUSDT": {19000: -0.4}} - first = _FakeBybit(panels) - ingest_bybit_worst_basis(store, first, ["AUSDT", "BUSDT"]) - assert set(first.calls) == {"AUSDT", "BUSDT"} + ingest_bybit_worst_basis(store, _FakeBybit({"AUSDT": {19000: -0.1}, "BUSDT": {19000: -0.2}}), + ["AUSDT", "BUSDT"]) + panels = {"AUSDT": {19000: -0.1, 19001: -0.15}, "BUSDT": {19000: -0.2, 19001: -0.25}, + "CUSDT": {19000: -0.3}, "DUSDT": {19000: -0.4}} second = _FakeBybit(panels) seen: list[str] = [] - summary = ingest_bybit_worst_basis( + ingest_bybit_worst_basis( store, second, ["AUSDT", "BUSDT", "CUSDT", "DUSDT"], resume=True, progress=lambda m: seen.append(m)) - assert set(second.calls) == {"CUSDT", "DUSDT"}, "resume re-fetched done symbols" - assert summary["symbols"] == 2 - assert any("resuming: 2 already done, 2 to go" in m for m in seen) + fetched = dict(second.calls) + assert set(fetched) == {"AUSDT", "BUSDT", "CUSDT", "DUSDT"} + assert fetched["AUSDT"] == 19000 * 86_400_000 and fetched["BUSDT"] == 19000 * 86_400_000 + assert fetched["CUSDT"] is None and fetched["DUSDT"] is None + assert any("resuming from last persisted date" in m for m in seen) - # final store holds all four + # done symbols ADVANCED to the new day; the rest were back-filled. full = store.read_panel(["AUSDT", "BUSDT", "CUSDT", "DUSDT"], "worst_basis") assert full == panels store.close() @@ -82,7 +86,7 @@ def test_no_resume_refetches_everything() -> None: ingest_bybit_worst_basis(store, _FakeBybit(panels), ["AUSDT"]) second = _FakeBybit(panels) ingest_bybit_worst_basis(store, second, ["AUSDT"], resume=False) - assert second.calls == ["AUSDT"] + assert second.calls == [("AUSDT", None)] store.close() @@ -92,7 +96,7 @@ def test_concurrent_ingest_writes_all_symbols_exactly_once() -> None: bybit = _FakeBybit(panels) summary = ingest_bybit_worst_basis(store, bybit, list(panels), workers=8) assert summary["symbols"] == 25 - assert sorted(bybit.calls) == sorted(panels) # each symbol fetched exactly once + assert sorted(c[0] for c in bybit.calls) == sorted(panels) # each symbol fetched exactly once assert store.read_panel(list(panels), "worst_basis") == panels store.close()