From b71cd6b3b0579b1d2340deac53e27009d1c101de Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 23:12:07 +0000 Subject: [PATCH] refactor(vrp): delete DVOL pipeline (deribit_dvol, dvol_ingest, VolIndexClient port) + tests --- src/fxhnt/adapters/data/deribit_dvol.py | 134 --------------------- src/fxhnt/application/dvol_ingest.py | 46 ------- src/fxhnt/ports/market_data.py | 6 - tests/integration/test_deribit_dvol.py | 152 ------------------------ tests/integration/test_dvol_ingest.py | 79 ------------ 5 files changed, 417 deletions(-) delete mode 100644 src/fxhnt/adapters/data/deribit_dvol.py delete mode 100644 src/fxhnt/application/dvol_ingest.py delete mode 100644 tests/integration/test_deribit_dvol.py delete mode 100644 tests/integration/test_dvol_ingest.py diff --git a/src/fxhnt/adapters/data/deribit_dvol.py b/src/fxhnt/adapters/data/deribit_dvol.py deleted file mode 100644 index f54ccdc..0000000 --- a/src/fxhnt/adapters/data/deribit_dvol.py +++ /dev/null @@ -1,134 +0,0 @@ -"""Deribit DVOL adapter — the DVOL annualised IMPLIED-vol index (the IV leg of the VRP edge). - -Why this signal: Deribit's DVOL is a BTC/ETH 30-day implied-volatility index (annualised, in vol points, e.g. -33–83 for BTC) computed off the live options order book — the market's forward-looking expected vol. Paired -with the REALIZED vol of the underlying it is the volatility risk premium (VRP): IV systematically exceeds -RV, so SELLING variance earns the spread, but with fat left tails when a vol spike makes RV >> IV. The -backtest SIGNAL uses DVOL (IV) + close-to-close RV; live execution would be short Bybit option straddles -(a documented cross-venue caveat — the DVOL index is Deribit's, the harvest is Bybit's options). - -Endpoint (HTTP is INJECTED so tests run with no network): - GET /api/v2/public/get_volatility_index_data?currency=BTC&start_timestamp=&end_timestamp= - &resolution=86400 - -> result.data[] of rows [ts_ms, open, high, low, close], oldest-first. close (idx 4) = the DVOL value. - resolution=86400 (seconds) = daily, so there is one row per UTC day. The API caps ~1000 rows/request, so - we paginate BACKWARD via end_timestamp for the full ~5.3y of history (BTC/ETH DVOL begins 2021-03-24). - -`dvol_history(currency, *, start_ms=None) -> {epoch_day: dvol_close}` paginates BACKWARD via end_timestamp -(mirrors BybitAccountRatio.long_ratio_history's shape) down to a floor (default ≈ 2021-03, just below the -DVOL inception 2021-03-24), stopping on the first empty page or once the oldest row falls below the floor. -epoch_day = -timestamp_ms // 86_400_000. - -CLOCK SEAM (mirrors DeribitFunding): the wall clock is INJECTED (`clock: Callable[[], float] = time.time`). -The first page anchors its end_timestamp at `now`; a bare `time.time()` in the loop would make the backward -walk depend on the calendar date the test runs on — the determinism bug this seam exists to prevent. Tests -pass a fixed clock so the walk is reproducible. -""" -from __future__ import annotations - -import datetime as _dt -import json -import time -import urllib.request -from collections.abc import Callable -from typing import Any - -_API = "https://www.deribit.com/api/v2/public/get_volatility_index_data" -_PAGE_LIMIT = 1000 # Deribit caps ~1000 rows/request on the volatility-index endpoint -_DAY_MS = 86_400_000 -_CLOSE_IX = 4 # row layout: [ts_ms, open, high, low, close] - -# Backstop floor for the backward pagination: ≈ 2021-03-01 UTC in ms. The Deribit DVOL index for BTC and ETH -# actually begins 2021-03-24 (verified against the live get_volatility_index_data endpoint); 2021-03-01 is the -# natural lower bound just below inception at which the backward walk terminates. (The previous 2023-09 floor -# silently discarded ~890 days — nearly half — of valid DVOL history, halving the VRP study's non-overlapping -# hold count; this floor recovers the full ~5.3y back to inception.) -DERIBIT_DVOL_FLOOR_MS = int(_dt.datetime(2021, 3, 1, tzinfo=_dt.UTC).timestamp() * 1000) - - -def _http_get_json(url: str, tries: int = 4) -> Any: - """Default JSON fetcher (urllib, retry+backoff) — mirrors the other free-data adapters' `_get`.""" - for a in range(tries): - try: - req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) - with urllib.request.urlopen(req, timeout=40) as r: - return json.loads(r.read()) - except Exception: # noqa: BLE001 — transient HTTP; retry then re-raise - if a == tries - 1: - raise - time.sleep(2 * (a + 1)) - raise RuntimeError("unreachable") - - -def _with_backoff(fetch: Callable[[str], Any], *, sleep: Callable[[float], None], - base: float, tries: int) -> Callable[[str], Any]: - """Wrap `fetch(url)->dict` with exponential backoff on ANY transient exception (the Deribit endpoint has - no Bybit-style retCode; a 429/5xx surfaces as a raised exception). Retries `base·2**attempt` up to - `tries`; the final attempt re-raises. `sleep` is injectable (tests pass a no-op recorder).""" - def wrapped(url: str) -> Any: - for attempt in range(tries): - try: - return fetch(url) - except Exception: # noqa: BLE001 — transient HTTP; backoff then re-raise on exhaustion - if attempt == tries - 1: - raise - sleep(base * (2 ** attempt)) - raise RuntimeError("unreachable") # pragma: no cover — loop always returns or raises - return wrapped - - -class DeribitDVOL: - def __init__(self, *, fetch: Callable[[str], Any] = _http_get_json, - sleep: Callable[[float], None] = time.sleep, - clock: Callable[[], float] = time.time, - backoff_base: float = 0.5, backoff_tries: int = 5) -> None: - # fetch(url) -> parsed JSON dict (injected for tests). Wrapped with exponential backoff so a transient - # 429/5xx from Deribit is retried instead of failing — `sleep` is the (injectable) backoff sleeper - # too, so tests passing a no-op sleep never actually wait. - self._fetch = _with_backoff(fetch, sleep=sleep, base=backoff_base, tries=backoff_tries) - self._sleep = sleep # throttle between pages in production; no-op-able in tests - # `clock` is the wall-clock seam (epoch seconds; defaults to time.time). The first page anchors its - # end_timestamp at `now`; injecting a FIXED clock makes the backward walk deterministic so tests do - # not depend on the calendar date they run on (mirrors DeribitFunding — a bare time.time() in the loop - # is the determinism bug this seam exists to prevent). - self._clock = clock - - def dvol_history(self, currency: str, *, start_ms: int | None = None) -> dict[int, float]: - """{epoch_day: dvol_close} for `currency` (BTC/ETH), where dvol_close = the daily DVOL index value - (annualised implied vol, vol points). resolution=86400 → one row/day, no aggregation. - - Paginates BACKWARD via end_timestamp: fetch the newest page, then set end_timestamp = oldest row's - ts − 1 and loop, stopping on the first empty page or when the oldest row falls below `start_ms` - (default DERIBIT_DVOL_FLOOR_MS ≈ 2021-03). Last-write-wins on a duplicate (currency, day) — there is - at most one row per day, so this only guards page-boundary repeats. The infinite-loop guard breaks if - the cursor stops advancing.""" - floor = DERIBIT_DVOL_FLOOR_MS if start_ms is None else int(start_ms) - daily: dict[int, float] = {} - seen: set[int] = set() # guard against duplicate ts across page boundaries - end_ms: int | None = None # first page: no cursor (newest) - while True: - now_ms = int(self._clock() * 1000) - url = (f"{_API}?currency={currency}&start_timestamp={floor}" - f"&end_timestamp={end_ms if end_ms is not None else now_ms}&resolution=86400") - data = self._fetch(url) - rows = data.get("result", {}).get("data", []) - if not rows: - break - oldest_ms = end_ms if end_ms is not None else None - for row in rows: - ts_ms = int(row[0]) - if oldest_ms is None or ts_ms < oldest_ms: - oldest_ms = ts_ms - if ts_ms < floor or ts_ms in seen: - continue - seen.add(ts_ms) - daily[ts_ms // _DAY_MS] = float(row[_CLOSE_IX]) - if oldest_ms is None or oldest_ms <= floor: - break # walked back to the floor → done - next_end = oldest_ms - 1 - if end_ms is not None and next_end >= end_ms: - break # cursor not advancing → avoid an infinite loop - end_ms = next_end - self._sleep(0.1) # gentle throttle; tests inject a no-op sleep - return daily diff --git a/src/fxhnt/application/dvol_ingest.py b/src/fxhnt/application/dvol_ingest.py deleted file mode 100644 index 54a8ae2..0000000 --- a/src/fxhnt/application/dvol_ingest.py +++ /dev/null @@ -1,46 +0,0 @@ -"""Ingest the Deribit DVOL implied-vol index into the Bybit-namespaced feature store (table='bybit_features') -as feature 'dvol' for the perp symbols BTCUSDT / ETHUSDT. - -The DVOL index is currency-keyed (BTC/ETH); the warehouse panels are symbol-keyed (the VRP eval reads `dvol` -alongside the BTCUSDT/ETHUSDT `close`), so the ingest maps currency -> "USDT" and writes feature='dvol' -at ts = epoch_day*86400. ONLY 'dvol' is written — close/funding are the kline/funding ingests' job (the PK is -(feature, symbol, ts) so distinct features never collide). Idempotent (upsert overwrites in place, so re-runs -are safe). Per-currency skip-on-empty: a currency that returns no rows contributes nothing and never aborts. -""" -from __future__ import annotations - -from typing import Any, Protocol - -# Deribit DVOL currency -> the Bybit USDT-perp symbol the warehouse panels key on. -_CCY_TO_SYMBOL = {"BTC": "BTCUSDT", "ETH": "ETHUSDT"} - - -class _Deribit(Protocol): - def dvol_history(self, currency: str, *, start_ms: int | None = None) -> dict[int, float]: ... - - -def ingest_dvol(store: Any, deribit: _Deribit, currencies: tuple[str, ...] = ("BTC", "ETH"), *, - start_ms: int | None = None) -> dict[str, Any]: - """Fetch + write the daily DVOL index for each `currency` into `store` (feature='dvol', - ts=epoch_day*86400, symbol="USDT"). - - Returns {"symbols": N, "day_rows": M, "oldest_day": D|None}: N = currencies that produced >=1 day-row, - M = total day-rows written, D = the earliest epoch_day across all currencies (None if nothing ingested). - Per-currency skip-on-empty (a currency with no DVOL history contributes nothing, never aborts). Writes - are bulk (one transaction); the upsert is idempotent so re-running is safe and cheap.""" - items: list[tuple[str, list[tuple[int, dict[str, float]]]]] = [] - oldest_day: int | None = None - day_rows = 0 - for ccy in currencies: - symbol = _CCY_TO_SYMBOL.get(ccy.upper(), f"{ccy.upper()}USDT") - daily = deribit.dvol_history(ccy, start_ms=start_ms) - if not daily: - continue - rows = [(day * 86_400, {"dvol": float(val)}) for day, val in sorted(daily.items())] - items.append((symbol, rows)) - day_rows += len(rows) - run_oldest = min(daily) - oldest_day = run_oldest if oldest_day is None else min(oldest_day, run_oldest) - if items: - store.write_features_bulk(items) - return {"symbols": len(items), "day_rows": day_rows, "oldest_day": oldest_day} diff --git a/src/fxhnt/ports/market_data.py b/src/fxhnt/ports/market_data.py index 8f10fce..37ca957 100644 --- a/src/fxhnt/ports/market_data.py +++ b/src/fxhnt/ports/market_data.py @@ -18,12 +18,6 @@ class CryptoPerpBarClient(Protocol): ... -class VolIndexClient(Protocol): - def dvol(self, currency: str) -> dict[int, float]: - """{ epoch_day: implied-vol-index } for BTC/ETH (Deribit DVOL).""" - ... - - class FundamentalsClient(Protocol): def metrics(self, symbol: str) -> dict[str, float]: """Latest daily valuation metrics: {'peRatio','pbRatio','marketCap', ...} (most recent available).""" diff --git a/tests/integration/test_deribit_dvol.py b/tests/integration/test_deribit_dvol.py deleted file mode 100644 index 437797f..0000000 --- a/tests/integration/test_deribit_dvol.py +++ /dev/null @@ -1,152 +0,0 @@ -"""DeribitDVOL adapter — the Deribit DVOL annualised implied-vol index (the IV leg of the VRP edge). - -ALL HTTP is injected/mocked — no network in these tests. The adapter takes a `fetch(url)->dict` callable -(the JSON endpoint), so the tests drive it with canned paginated responses and assert: the {epoch_day: -dvol_close} mapping (close = row index 4), backward pagination via end_timestamp to a floor, the floor stop, -the empty-page stop, dedupe across page boundaries, and rate-limit/transient backoff. -""" -from __future__ import annotations - -import datetime as _dt - -from fxhnt.adapters.data.deribit_dvol import DERIBIT_DVOL_FLOOR_MS, DeribitDVOL - -_DAY_MS = 86_400_000 - -# DETERMINISM SEAM: pin the adapter's wall clock to a fixed recent instant so the backward walk anchors at a -# constant `now` regardless of the calendar date the suite runs on (mirrors test_deribit_funding). If the -# adapter ever reverts to a bare time.time() in the loop, the clock-determinism guard below fails. -_NOW_MS = 1780315200_000 # noon 2026-06-01 UTC, in ms - - -def _FIXED_CLOCK() -> float: # noqa: N802 — constant-like injectable clock - return _NOW_MS / 1000.0 - - -def _dvol_row(ts_ms: int, close: float) -> list: - # Deribit volatility-index rows: [ts_ms, open, high, low, close]. close (idx 4) is the DVOL value. - return [ts_ms, close - 1.0, close + 1.0, close - 2.0, close] - - -class _PagedDVOL: - """Mock of /public/get_volatility_index_data: serves rows oldest->newest within the requested - [start_timestamp, end_timestamp] window, honoring the end_timestamp cursor like Deribit (returns up to - `limit` rows ending at end_timestamp, oldest-first). Walking end_timestamp back past the oldest row - returns an empty list (stops the loop).""" - - def __init__(self, rows: list[list], *, limit: int = 1000) -> None: - self._rows = sorted(rows, key=lambda r: r[0]) - self._limit = limit - self.calls: list[tuple[int | None, int | None]] = [] - - def fetch(self, url: str) -> dict: - start = _parse_qs_int(url, "start_timestamp") - end = _parse_qs_int(url, "end_timestamp") - self.calls.append((start, end)) - cand = [r for r in self._rows - if (start is None or r[0] >= start) and (end is None or r[0] <= end)] - # newest `limit` rows in the window (Deribit caps 1000/request, returns the most recent of the window) - page = cand[-self._limit:] - return {"result": {"data": page}} - - -def _parse_qs_int(url: str, key: str) -> int | None: - from urllib.parse import parse_qs, urlparse - q = parse_qs(urlparse(url).query) - return int(q[key][0]) if key in q else None - - -def test_dvol_history_maps_epoch_day_to_close_and_paginates_backward(): - rows = [ - _dvol_row(100 * _DAY_MS, 60.0), - _dvol_row(99 * _DAY_MS, 55.0), - _dvol_row(98 * _DAY_MS, 48.0), - _dvol_row(97 * _DAY_MS, 52.0), - ] - paged = _PagedDVOL(rows, limit=2) - dv = DeribitDVOL(fetch=paged.fetch, sleep=lambda _: None) - daily = dv.dvol_history("BTC", start_ms=0) - assert daily == {100: 60.0, 99: 55.0, 98: 48.0, 97: 52.0} - assert len(paged.calls) >= 2 # it paginated and walked end_timestamp backward - - -def test_dvol_history_url_carries_currency_and_resolution(): - seen: list[str] = [] - - def fetch(url: str) -> dict: - seen.append(url) - return {"result": {"data": []}} - - dv = DeribitDVOL(fetch=fetch, sleep=lambda _: None) - dv.dvol_history("ETH") - assert "get_volatility_index_data" in seen[0] - assert "currency=ETH" in seen[0] - assert "resolution=86400" in seen[0] - - -def test_dvol_history_empty_page_stops_loop(): - paged = _PagedDVOL([], limit=1000) - dv = DeribitDVOL(fetch=paged.fetch, sleep=lambda _: None) - assert dv.dvol_history("BTC") == {} - assert len(paged.calls) == 1 - - -def test_dvol_history_respects_start_ms_floor(): - rows = [_dvol_row(10 * _DAY_MS, 50.0), _dvol_row(20 * _DAY_MS, 60.0)] - paged = _PagedDVOL(rows, limit=1000) - dv = DeribitDVOL(fetch=paged.fetch, sleep=lambda _: None) - daily = dv.dvol_history("BTC", start_ms=15 * _DAY_MS) - assert 20 in daily and 10 not in daily # only the row at/after the floor survives - - -def test_default_floor_is_2021_03(): - # The floor sits just below the true BTC/ETH DVOL inception (2021-03-24); the old 2023-09 floor silently - # discarded ~890 days of valid history. - import datetime as dt - assert int(dt.datetime(2021, 3, 1, tzinfo=dt.UTC).timestamp() * 1000) == DERIBIT_DVOL_FLOOR_MS - - -def test_dvol_history_dedups_across_page_boundaries(): - rows = [_dvol_row(d * _DAY_MS, 40.0 + d) for d in range(50, 55)] - paged = _PagedDVOL(rows, limit=2) - dv = DeribitDVOL(fetch=paged.fetch, sleep=lambda _: None) - daily = dv.dvol_history("BTC", start_ms=0) - assert set(daily) == {50, 51, 52, 53, 54} - - -def test_dvol_history_is_clock_deterministic(): - # The first page's end_timestamp anchors at the INJECTED clock, not the wall clock: the URL of the first - # request is identical across runs when a fixed clock is supplied — proving no bare time.time() leaks into - # the loop (the determinism bug the clock seam prevents). - seen_a: list[str] = [] - seen_b: list[str] = [] - DeribitDVOL(fetch=lambda u: (seen_a.append(u), {"result": {"data": []}})[1], - sleep=lambda _: None, clock=_FIXED_CLOCK).dvol_history("BTC") - DeribitDVOL(fetch=lambda u: (seen_b.append(u), {"result": {"data": []}})[1], - sleep=lambda _: None, clock=_FIXED_CLOCK).dvol_history("BTC") - assert seen_a == seen_b - assert f"end_timestamp={_NOW_MS}" in seen_a[0] # anchored at the fixed clock, not today - assert int(_FIXED_CLOCK()) == 1780315200 # frozen anchor, never tracks today - - -def test_default_floor_matches_2021_03(): - assert int(_dt.datetime(2021, 3, 1, tzinfo=_dt.UTC).timestamp() * 1000) == DERIBIT_DVOL_FLOOR_MS - - -def test_dvol_history_retries_on_transient_error_then_succeeds(): - d = 30 * _DAY_MS - real = _PagedDVOL([_dvol_row(d, 70.0)], limit=1000) - state = {"n": 0} - - def flaky(url: str) -> dict: - state["n"] += 1 - if state["n"] <= 2: - raise RuntimeError("transient 502") - return real.fetch(url) - - waited: list[float] = [] - dv = DeribitDVOL(fetch=flaky, sleep=waited.append, backoff_base=0.5, backoff_tries=5) - daily = dv.dvol_history("BTC", start_ms=0) - assert abs(daily[30] - 70.0) < 1e-12 - assert state["n"] >= 3 - assert waited[:2] == [0.5, 1.0] # exponential backoff (no real sleep) diff --git a/tests/integration/test_dvol_ingest.py b/tests/integration/test_dvol_ingest.py deleted file mode 100644 index 3d3b9c9..0000000 --- a/tests/integration/test_dvol_ingest.py +++ /dev/null @@ -1,79 +0,0 @@ -"""ingest_dvol — write the Deribit DVOL index into bybit_features as feature 'dvol' for BTCUSDT/ETHUSDT. - -The DVOL index is currency-keyed (BTC/ETH); the warehouse panels are symbol-keyed (BTCUSDT/ETHUSDT), so the -ingest maps currency -> symbol and writes feature='dvol' at ts = epoch_day*86400. ONLY 'dvol' is written -(close is the kline ingest's job). Idempotent/resumable. No network — the adapter is a stub. -""" -from __future__ import annotations - -from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore -from fxhnt.application.dvol_ingest import ingest_dvol - -_DAY = 86_400 - - -class _StubDeribit: - """Stub DeribitDVOL: serves a canned {epoch_day: dvol} per currency, records the calls.""" - - def __init__(self, by_currency: dict[str, dict[int, float]]) -> None: - self._by_currency = by_currency - self.calls: list[str] = [] - - def dvol_history(self, currency: str, *, start_ms: int | None = None) -> dict[int, float]: - self.calls.append(currency) - return dict(self._by_currency.get(currency, {})) - - -def test_ingest_writes_dvol_for_btc_and_eth() -> None: - store = TimescaleFeatureStore("sqlite://", table="bybit_features") - deribit = _StubDeribit({"BTC": {100: 55.0, 101: 60.0}, "ETH": {100: 70.0, 101: 72.0}}) - summary = ingest_dvol(store, deribit) - # BTCUSDT/ETHUSDT now carry the 'dvol' feature at the right timestamps. - btc = store.read_panel(["BTCUSDT"], "dvol")["BTCUSDT"] - eth = store.read_panel(["ETHUSDT"], "dvol")["ETHUSDT"] - store.close() - assert btc == {100: 55.0, 101: 60.0} - assert eth == {100: 70.0, 101: 72.0} - assert summary["symbols"] == 2 - assert summary["day_rows"] == 4 - assert summary["oldest_day"] == 100 - assert set(deribit.calls) == {"BTC", "ETH"} - - -def test_ingest_maps_currency_to_usdt_symbol() -> None: - store = TimescaleFeatureStore("sqlite://", table="bybit_features") - deribit = _StubDeribit({"BTC": {1: 50.0}}) - ingest_dvol(store, deribit, currencies=("BTC",)) - # The row lands under BTCUSDT at ts = epoch_day*86400, not under "BTC". - rows = store.read_features("BTCUSDT", features=["dvol"]) - store.close() - assert rows == [(1 * _DAY, {"dvol": 50.0})] - - -def test_ingest_skips_currency_with_no_data() -> None: - store = TimescaleFeatureStore("sqlite://", table="bybit_features") - deribit = _StubDeribit({"BTC": {1: 50.0}}) # ETH returns nothing - summary = ingest_dvol(store, deribit) - store.close() - assert summary["symbols"] == 1 # only BTC produced rows; ETH skipped (never aborts) - - -def test_ingest_is_idempotent() -> None: - store = TimescaleFeatureStore("sqlite://", table="bybit_features") - deribit = _StubDeribit({"BTC": {1: 50.0}, "ETH": {1: 60.0}}) - ingest_dvol(store, deribit) - ingest_dvol(store, deribit) # re-run overwrites in place, no duplicate rows - btc = store.read_features("BTCUSDT", features=["dvol"]) - store.close() - assert btc == [(1 * _DAY, {"dvol": 50.0})] - - -def test_ingest_only_writes_dvol_feature() -> None: - store = TimescaleFeatureStore("sqlite://", table="bybit_features") - deribit = _StubDeribit({"BTC": {1: 50.0}}) - ingest_dvol(store, deribit, currencies=("BTC",)) - rows = store.read_features("BTCUSDT") - store.close() - # exactly one feature written, and it's 'dvol' (close/funding are the other ingests' job). - feats = {name for _, feats in rows for name in feats} - assert feats == {"dvol"}