gitadmin/fxhnt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ingest): date-resume all bybit per-symbol ingests so funding/oi/long_ratio/worst_basis advance nightly

Browse Source 
The spot fix introduced `last_day_by_symbol` date-resume, but the sibling bybit ingests still used the
OLD symbol-skip resume ("resuming: N already done, M to go"): once a symbol carried ANY row of a feature
it was never fetched again, so the feature froze at the last ingest. funding stuck at 2026-06-23 froze
`xvenue_carry_nav` (it reads bybit_features funding) at 0d; open_interest/long_ratio/worst_basis lagged.

Extract a shared `resume_floor_ms(last_day, symbol, default_start_ms)` primitive (bybit_concurrency) and
apply it to `ingest_bybit_funding`, `ingest_bybit_oi`, `ingest_bybit_long_short`, and
`ingest_bybit_worst_basis`: each symbol now resumes from its last persisted day for ITS feature (the
adapter's backward-pagination floor), so done symbols are re-fetched only from their last day forward
(advancing them; the idempotent upsert overwrites the boundary day in place) and fresh symbols back-fill
from the caller's start_ms. Nothing is symbol-skipped → every feature advances each run. Refactored
`ingest_bybit_spot` onto the same shared primitive. deribit_funding already re-fetches all (no skip) — left
as-is. The perp `close`/`turnover` leg of `ingest_bybit_klines` keeps its symbol-skip resume (out of scope).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-06-28 11:29:35 +02:00
parent 30e723c89e
commit ad5b9dcc28
11 changed files with 202 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/fxhnt/application/bybit_concurrency.py
View File

@@ -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."""

35
src/fxhnt/application/bybit_funding_ingest.py
View File

@@ -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}

11
src/fxhnt/application/bybit_klines_ingest.py
View File

@@ -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

37
src/fxhnt/application/bybit_long_short_ingest.py
View File

@@ -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}

37
src/fxhnt/application/bybit_oi_ingest.py
View File

@@ -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}

33
src/fxhnt/application/bybit_worst_basis_ingest.py
View File

@@ -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}

18
tests/integration/test_bybit_forward_track.py
View File

@@ -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

51
tests/integration/test_bybit_funding_ingest.py
View File

@@ -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()

37
tests/integration/test_bybit_long_short_ingest.py
View File

@@ -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()

36
tests/integration/test_bybit_oi_ingest.py
View File

@@ -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()

38
tests/integration/test_bybit_worst_basis_ingest.py
View File

@@ -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()