298 lines
13 KiB
Python
298 lines
13 KiB
Python
"""READ-ONLY evaluator for the Bybit OI-DELEVERAGING REVERSION edge (open-interest unwind overshoot).
|
|
|
|
Signal (per coin/day): on an OI-DROP day (doi<0 = forced unwind) FADE that day's move with magnitude ∝
|
|
|doi|·|ret|; on OI-build days (doi>=0) the signal is zero. Cross-sectional, market-neutral, unit-gross
|
|
weights bet on the NEXT-day price return (causal). REVERSION (default) fades; MOMENTUM is the sign-flip.
|
|
|
|
Tests assert:
|
|
* doi/ret are computed day-over-day, keyed by the later day;
|
|
* the deleveraging-overshoot formula: fade on OI-DROP days, zero on OI-build days, momentum = negation;
|
|
* weights are market-neutral (Σw≈0) + unit-gross (Σ|w|≈1);
|
|
* on a fixture where fading forced unwinds is profitable, reversion Sharpe > 0 and momentum < 0;
|
|
* the evaluator is READ-ONLY (a write-tripwire never trips);
|
|
* the verify report is cost-monotone, per-coin sums to total, slices per-year, and carries correlation
|
|
fields vs ALL FOUR existing edges (tstrend, unlock, xsfunding, positioning);
|
|
* the CLI smoke prints the metrics + verify diagnostics (mocked in-memory store, NO network).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
|
|
from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
|
|
from fxhnt.application.bybit_oi_deleverage_eval import (
|
|
deleverage_signal_by_day,
|
|
deleverage_weights,
|
|
oi_deleverage_metrics_from_store,
|
|
verify_oi_deleverage_edge,
|
|
)
|
|
|
|
_DAY = 86_400
|
|
|
|
|
|
class _WriteTripwireStore:
|
|
"""Read-only proxy: forwards reads, but ANY write/persist call trips an assertion."""
|
|
|
|
_FORBIDDEN = frozenset({
|
|
"write_features", "write_features_bulk", "upsert_feature_rows", "upsert_membership",
|
|
"replace_positions", "upsert_nav", "upsert_sleeve_ret", "replace_shadow_positions",
|
|
"replace_trades", "_create_schema", "_bulk_upsert",
|
|
})
|
|
|
|
def __init__(self, inner: TimescaleFeatureStore) -> None:
|
|
object.__setattr__(self, "_inner", inner)
|
|
|
|
def __getattr__(self, name: str):
|
|
if name in _WriteTripwireStore._FORBIDDEN:
|
|
raise AssertionError(f"READ-ONLY violation: evaluator called write method {name!r}")
|
|
return getattr(self._inner, name)
|
|
|
|
|
|
# --- 1. the deleveraging signal ----------------------------------------------------------------
|
|
|
|
def test_signal_fades_move_on_oi_drop_day() -> None:
|
|
# AAA: OI dropped (forced unwind) and price went UP that day -> fade -> SHORT (negative signal).
|
|
# BBB: OI dropped and price went DOWN that day -> fade -> LONG (positive signal).
|
|
doi = {1: {"AAAUSDT": -0.20, "BBBUSDT": -0.20}}
|
|
ret = {1: {"AAAUSDT": +0.05, "BBBUSDT": -0.05}}
|
|
sig = deleverage_signal_by_day(doi, ret)[1]
|
|
assert sig["AAAUSDT"] < 0.0 # up-on-unwind -> fade down (short)
|
|
assert sig["BBBUSDT"] > 0.0 # down-on-unwind -> fade up (long)
|
|
|
|
|
|
def test_signal_zero_on_oi_build_day() -> None:
|
|
# OI BUILDING (doi >= 0) is fresh positioning, not a forced unwind -> no signal.
|
|
doi = {1: {"AAAUSDT": +0.10, "BBBUSDT": 0.0}}
|
|
ret = {1: {"AAAUSDT": +0.05, "BBBUSDT": -0.05}}
|
|
sig = deleverage_signal_by_day(doi, ret)[1]
|
|
assert sig["AAAUSDT"] == 0.0
|
|
assert sig["BBBUSDT"] == 0.0
|
|
|
|
|
|
def test_signal_magnitude_scales_with_doi_and_ret() -> None:
|
|
# bigger unwind AND bigger move => stronger overshoot signal. Both rets POSITIVE here, so the formula
|
|
# signal = -sign(ret)*|doi|*|ret| = -1 * |doi| * |ret| (the fade is a SHORT).
|
|
doi = {1: {"SMALLUSDT": -0.10, "BIGUSDT": -0.40}}
|
|
ret = {1: {"SMALLUSDT": +0.02, "BIGUSDT": +0.08}}
|
|
sig = deleverage_signal_by_day(doi, ret)[1]
|
|
assert abs(sig["BIGUSDT"]) > abs(sig["SMALLUSDT"])
|
|
assert math.isclose(sig["BIGUSDT"], -1.0 * 0.40 * 0.08, abs_tol=1e-12)
|
|
assert math.isclose(sig["SMALLUSDT"], -1.0 * 0.10 * 0.02, abs_tol=1e-12)
|
|
|
|
|
|
def test_momentum_signal_is_exact_negation() -> None:
|
|
doi = {1: {"AAAUSDT": -0.20, "BBBUSDT": -0.30}}
|
|
ret = {1: {"AAAUSDT": +0.05, "BBBUSDT": -0.05}}
|
|
rev = deleverage_signal_by_day(doi, ret, direction="reversion")[1]
|
|
mom = deleverage_signal_by_day(doi, ret, direction="momentum")[1]
|
|
for c in rev:
|
|
assert math.isclose(mom[c], -rev[c], abs_tol=1e-12)
|
|
|
|
|
|
def test_weights_market_neutral_and_unit_gross() -> None:
|
|
sig = {1: {"AAAUSDT": -0.04, "BBBUSDT": 0.02, "CCCUSDT": 0.01, "DDDUSDT": -0.03}}
|
|
w = deleverage_weights(sig)[1]
|
|
assert abs(sum(w.values())) < 1e-9
|
|
assert math.isclose(sum(abs(v) for v in w.values()), 1.0, abs_tol=1e-9)
|
|
|
|
|
|
def test_weights_skip_all_zero_signal_day() -> None:
|
|
# an all-OI-build day demeans to all-zero -> zero gross -> skipped.
|
|
sig = {0: {"AAAUSDT": 0.0, "BBBUSDT": 0.0}, 1: {"AAAUSDT": -0.02, "BBBUSDT": 0.0}}
|
|
out = deleverage_weights(sig)
|
|
assert 0 not in out and 1 in out
|
|
|
|
|
|
# --- 2. metrics --------------------------------------------------------------------------------
|
|
|
|
def _seed_reversion_clean(store: TimescaleFeatureStore, *, days: int = 150,
|
|
coins: tuple[str, ...] = ("AAAUSDT", "BBBUSDT", "CCCUSDT", "DDDUSDT"),
|
|
start_day: int = 18628) -> None:
|
|
"""Clean reversion fixture: each day ONE rotating coin has a forced OI unwind (OI drops) with a sharp
|
|
move; the FOLLOWING day that coin's price reverts (fade earns). Non-unwinding coins keep OI ~flat (no
|
|
signal) and prices ~flat. start near 2021-01-01 so a long span crosses calendar years."""
|
|
px = {c: 100.0 for c in coins}
|
|
oi = {c: 1_000_000.0 for c in coins}
|
|
pending_revert: dict[str, float] = {} # coin -> the next-day multiplicative reversion to apply
|
|
for c in coins:
|
|
store.write_features(c, [(start_day * _DAY, {"open_interest": oi[c], "close": px[c]})])
|
|
for i in range(1, days + 1):
|
|
d = start_day + i
|
|
unwind = coins[i % len(coins)]
|
|
up = (i % 2 == 0)
|
|
for c in coins:
|
|
# First, apply any pending reversion from yesterday's overshoot (this is the move the fade earns).
|
|
if c in pending_revert:
|
|
px[c] *= pending_revert.pop(c)
|
|
if c == unwind:
|
|
oi[c] *= 0.78 # forced unwind: OI drops sharply
|
|
move = 0.06 if up else -0.06
|
|
px[c] *= (1.0 + move) # sharp overshoot today
|
|
pending_revert[c] = (1.0 - move * 0.9) # ~revert tomorrow (slightly less -> net edge)
|
|
else:
|
|
oi[c] *= 1.0005 # ~flat OI -> no forced-unwind signal
|
|
store.write_features(c, [(d * _DAY, {"open_interest": oi[c], "close": px[c]})])
|
|
# one trailing day to realize the final pending reversion as a next-day return.
|
|
d = start_day + days + 1
|
|
for c in coins:
|
|
if c in pending_revert:
|
|
px[c] *= pending_revert.pop(c)
|
|
store.write_features(c, [(d * _DAY, {"open_interest": oi[c], "close": px[c]})])
|
|
|
|
|
|
def test_metrics_reversion_positive_when_fading_unwinds_pays() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_reversion_clean(store)
|
|
m = oi_deleverage_metrics_from_store(store, cost_bps=0.0, direction="reversion")
|
|
store.close()
|
|
assert m["available"] is True
|
|
assert m["n_coins"] == 4
|
|
assert m["days"] > 0
|
|
assert math.isfinite(m["sharpe"])
|
|
assert m["total_return"] > 0.0
|
|
|
|
|
|
def test_metrics_direction_flip_makes_momentum_lose() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_reversion_clean(store)
|
|
m_rev = oi_deleverage_metrics_from_store(store, cost_bps=0.0, direction="reversion")
|
|
m_mom = oi_deleverage_metrics_from_store(store, cost_bps=0.0, direction="momentum")
|
|
store.close()
|
|
assert m_mom["sharpe"] < 0.0 < m_rev["sharpe"]
|
|
|
|
|
|
def test_metrics_na_when_no_oi_data() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
store.write_features("AAAUSDT", [(0, {"close": 100.0})]) # close but no open_interest
|
|
m = oi_deleverage_metrics_from_store(store)
|
|
store.close()
|
|
assert m["available"] is False
|
|
assert "reason" in m and m["reason"]
|
|
|
|
|
|
def test_metrics_is_read_only() -> None:
|
|
inner = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_reversion_clean(inner)
|
|
store = _WriteTripwireStore(inner)
|
|
m = oi_deleverage_metrics_from_store(store, cost_bps=5.5) # must not trip the tripwire
|
|
inner.close()
|
|
assert m["available"] is True
|
|
|
|
|
|
# --- 3. verify diagnostics ---------------------------------------------------------------------
|
|
|
|
def test_verify_reversion_beats_momentum_on_reversion_fixture() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_reversion_clean(store)
|
|
rep = verify_oi_deleverage_edge(store, cost_bps=5.5)
|
|
store.close()
|
|
assert rep["available"] is True
|
|
rev = rep["net_cost"]["reversion"][5.5]
|
|
mom = rep["net_cost"]["momentum"][5.5]
|
|
assert rev["sharpe"] > 0.0 > mom["sharpe"]
|
|
|
|
|
|
def test_verify_cost_sensitivity_is_monotone() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_reversion_clean(store)
|
|
rep = verify_oi_deleverage_edge(store, cost_bps=5.5, cost_grid=(5.5, 11.0, 22.0))
|
|
store.close()
|
|
rev = rep["net_cost"]["reversion"]
|
|
assert rev[5.5]["sharpe"] >= rev[11.0]["sharpe"] >= rev[22.0]["sharpe"]
|
|
|
|
|
|
def test_verify_per_coin_attribution_sums_to_total() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_reversion_clean(store)
|
|
rep = verify_oi_deleverage_edge(store, cost_bps=0.0)
|
|
store.close()
|
|
attr = rep["per_coin"]["reversion"]
|
|
total = rep["gross_total"]["reversion"]
|
|
assert math.isclose(sum(attr.values()), total, rel_tol=1e-9, abs_tol=1e-12)
|
|
assert 0.0 <= rep["top5_share"]["reversion"] <= 1.0 + 1e-9
|
|
|
|
|
|
def test_verify_per_year_sharpes_on_right_slices() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_reversion_clean(store, days=420) # spans 2021 and 2022
|
|
rep = verify_oi_deleverage_edge(store, cost_bps=5.5)
|
|
store.close()
|
|
per_year = rep["per_year"]["reversion"]
|
|
assert "2021" in per_year and "2022" in per_year
|
|
total_days = rep["net_cost"]["reversion"][5.5]["days"]
|
|
assert sum(y["days"] for y in per_year.values()) == total_days
|
|
|
|
|
|
def test_verify_reports_turnover() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_reversion_clean(store)
|
|
rep = verify_oi_deleverage_edge(store, cost_bps=5.5)
|
|
store.close()
|
|
assert rep["avg_turnover"]["reversion"] > 0.0
|
|
assert rep["avg_turnover"]["momentum"] > 0.0
|
|
|
|
|
|
def test_verify_reports_correlation_with_all_four_edges() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_reversion_clean(store)
|
|
rep = verify_oi_deleverage_edge(store, cost_bps=5.5)
|
|
store.close()
|
|
for d in ("reversion", "momentum"):
|
|
corrs = rep["corr_edges"][d]
|
|
for edge in ("tstrend", "unlock", "xsfunding", "positioning"):
|
|
assert edge in corrs
|
|
v = corrs[edge]
|
|
assert v is None or (-1.0 - 1e-9 <= v <= 1.0 + 1e-9)
|
|
|
|
|
|
def test_verify_is_read_only() -> None:
|
|
inner = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_reversion_clean(inner)
|
|
store = _WriteTripwireStore(inner)
|
|
rep = verify_oi_deleverage_edge(store, cost_bps=5.5)
|
|
inner.close()
|
|
assert rep["available"] is True
|
|
|
|
|
|
# --- 4. CLI smoke ------------------------------------------------------------------------------
|
|
|
|
def test_cli_oi_deleverage_eval_prints_metrics(monkeypatch) -> None:
|
|
from typer.testing import CliRunner
|
|
|
|
import fxhnt.cli as cli
|
|
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_reversion_clean(store)
|
|
|
|
monkeypatch.setattr(
|
|
"fxhnt.adapters.warehouse.timescale_feature_store.TimescaleFeatureStore",
|
|
lambda *_a, **_k: store, raising=False)
|
|
|
|
result = CliRunner().invoke(cli.app, ["bybit-oi-deleverage-eval", "--all"])
|
|
store.close()
|
|
assert result.exit_code == 0, result.output
|
|
out = result.output.lower()
|
|
assert "oi-deleveraging" in out or "deleverag" in out
|
|
assert "reversion" in out
|
|
|
|
|
|
def test_cli_oi_deleverage_eval_verify_prints_diagnostics(monkeypatch) -> None:
|
|
from typer.testing import CliRunner
|
|
|
|
import fxhnt.cli as cli
|
|
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_reversion_clean(store)
|
|
|
|
monkeypatch.setattr(
|
|
"fxhnt.adapters.warehouse.timescale_feature_store.TimescaleFeatureStore",
|
|
lambda *_a, **_k: store, raising=False)
|
|
|
|
result = CliRunner().invoke(cli.app, ["bybit-oi-deleverage-eval", "--all", "--verify"])
|
|
store.close()
|
|
assert result.exit_code == 0, result.output
|
|
out = result.output.lower()
|
|
assert "reversion" in out and "momentum" in out
|
|
assert "per-coin" in out or "per coin" in out
|
|
assert "turnover" in out
|
|
assert "tstrend" in out and "unlock" in out and "xsfunding" in out and "positioning" in out
|