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

refactor(vrp): delete DVOL pipeline (deribit_dvol, dvol_ingest, VolIndexClient port) + tests

Browse Source
This commit is contained in:
Jeroen Grusewski
2026-07-20 23:12:07 +00:00
parent 59cc06e594
commit b71cd6b3b0
5 changed files with 0 additions and 417 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
src/fxhnt/adapters/data/deribit_dvol.py
View File

@@ -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=<ms>&end_timestamp=<ms>
&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

46
src/fxhnt/application/dvol_ingest.py
View File

@@ -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 -> "<CCY>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="<CCY>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}

6
src/fxhnt/ports/market_data.py
View File

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

152
tests/integration/test_deribit_dvol.py
View File

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

79
tests/integration/test_dvol_ingest.py
View File

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