321 lines
13 KiB
Python
321 lines
13 KiB
Python
"""OOS / WALK-FORWARD validation harness for the Bybit OI-DELEVERAGING edge — the DEPLOY GATE.
|
|
|
|
The signal (forced OI-unwind overshoot) is near-parameterless, so OOS validation is about:
|
|
1. CONSISTENCY across held-out time -> rolling non-overlapping windows, fraction positive;
|
|
2. the DIRECTION CHOICE survives a real train/test split (re-pick on TRAIN, apply to never-seen TEST);
|
|
3. ROBUSTNESS -> drop-top-N contributors, liquidity sweep;
|
|
4. STRICT CAUSALITY -> a look-ahead audit (signal at d -> d->d+1 return only).
|
|
|
|
All fixtures are in-memory (sqlite), NO network, READ-ONLY (a write-tripwire never trips), memory-bounded.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
|
|
from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
|
|
from fxhnt.application.bybit_oi_deleverage_eval import (
|
|
look_ahead_audit,
|
|
walk_forward_oi_deleverage,
|
|
)
|
|
|
|
_DAY = 86_400
|
|
_START = 19_358 # 2023-01-01 (epoch day)
|
|
|
|
|
|
class _WriteTripwireStore:
|
|
_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: harness called write method {name!r}")
|
|
return getattr(self._inner, name)
|
|
|
|
|
|
# --- fixtures ----------------------------------------------------------------------------------
|
|
|
|
def _seed_broad_reversion(store: TimescaleFeatureStore, *, days: int = 720, revert_frac: float = 0.9,
|
|
coins: tuple[str, ...] = ("AAAUSDT", "BBBUSDT", "CCCUSDT", "DDDUSDT",
|
|
"EEEUSDT", "FFFUSDT", "GGGUSDT", "HHHUSDT"),
|
|
start_day: int = _START) -> None:
|
|
"""Broad, consistently-positive reversion edge across many coins for the FULL span: each day ONE rotating
|
|
coin has a forced OI unwind (OI drops) + a sharp move that REVERTS next day (the fade earns). Non-unwind
|
|
coins keep OI flat (no signal). `revert_frac` < 1 leaves a net edge after the bounce."""
|
|
px = {c: 100.0 for c in coins}
|
|
oi = {c: 1_000_000.0 for c in coins}
|
|
pending: dict[str, float] = {}
|
|
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:
|
|
if c in pending:
|
|
px[c] *= pending.pop(c)
|
|
if c == unwind:
|
|
oi[c] *= 0.78
|
|
move = 0.06 if up else -0.06
|
|
px[c] *= (1.0 + move)
|
|
pending[c] = (1.0 - move * revert_frac)
|
|
else:
|
|
oi[c] *= 1.0005
|
|
store.write_features(c, [(d * _DAY, {"open_interest": oi[c], "close": px[c]})])
|
|
d = start_day + days + 1
|
|
for c in coins:
|
|
if c in pending:
|
|
px[c] *= pending.pop(c)
|
|
store.write_features(c, [(d * _DAY, {"open_interest": oi[c], "close": px[c]})])
|
|
|
|
|
|
def _seed_train_only(store: TimescaleFeatureStore, *, days: int = 600) -> None:
|
|
"""Reversion pays in TRAIN (first 60%) but the sign FLIPS in TEST (forced moves CONTINUE there). Train
|
|
picks 'reversion'; applied to TEST it LOSES — the train/holdout test must catch the overfit."""
|
|
coins = ("AAAUSDT", "BBBUSDT", "CCCUSDT", "DDDUSDT")
|
|
px = {c: 100.0 for c in coins}
|
|
oi = {c: 1_000_000.0 for c in coins}
|
|
split = int(days * 0.6)
|
|
pending: dict[str, float] = {}
|
|
for c in coins:
|
|
store.write_features(c, [(_START * _DAY, {"open_interest": oi[c], "close": px[c]})])
|
|
for i in range(1, days + 1):
|
|
d = _START + i
|
|
unwind = coins[i % len(coins)]
|
|
up = (i % 2 == 0)
|
|
for c in coins:
|
|
if c in pending:
|
|
px[c] *= pending.pop(c)
|
|
if c == unwind:
|
|
oi[c] *= 0.78
|
|
move = 0.06 if up else -0.06
|
|
px[c] *= (1.0 + move)
|
|
# TRAIN: revert (fade earns). TEST: continue (forced move keeps going -> fade loses).
|
|
pending[c] = (1.0 - move * 0.9) if i < split else (1.0 + move * 0.9)
|
|
else:
|
|
oi[c] *= 1.0005
|
|
store.write_features(c, [(d * _DAY, {"open_interest": oi[c], "close": px[c]})])
|
|
d = _START + days + 1
|
|
for c in coins:
|
|
if c in pending:
|
|
px[c] *= pending.pop(c)
|
|
store.write_features(c, [(d * _DAY, {"open_interest": oi[c], "close": px[c]})])
|
|
|
|
|
|
def _seed_one_coin(store: TimescaleFeatureStore, *, days: int = 400) -> None:
|
|
"""The reversion edge lives in ONE coin only: AAA unwinds + reverts (tradeable); the others NEVER unwind
|
|
(OI flat -> no signal). Dropping the top contributor should KILL it (concentration-driven / fragile)."""
|
|
coins = ("AAAUSDT", "BBBUSDT", "CCCUSDT", "DDDUSDT")
|
|
px = {c: 100.0 for c in coins}
|
|
oi = {c: 1_000_000.0 for c in coins}
|
|
pending: dict[str, float] = {}
|
|
for c in coins:
|
|
store.write_features(c, [(_START * _DAY, {"open_interest": oi[c], "close": px[c]})])
|
|
for i in range(1, days + 1):
|
|
d = _START + i
|
|
up = (i % 2 == 0)
|
|
for c in coins:
|
|
if c in pending:
|
|
px[c] *= pending.pop(c)
|
|
if c == "AAAUSDT":
|
|
oi[c] *= 0.78
|
|
move = 0.06 if up else -0.06
|
|
px[c] *= (1.0 + move)
|
|
pending[c] = (1.0 - move * 0.9)
|
|
else:
|
|
oi[c] *= 1.0005 # others never unwind -> no signal, flat price
|
|
store.write_features(c, [(d * _DAY, {"open_interest": oi[c], "close": px[c]})])
|
|
d = _START + days + 1
|
|
for c in coins:
|
|
if c in pending:
|
|
px[c] *= pending.pop(c)
|
|
store.write_features(c, [(d * _DAY, {"open_interest": oi[c], "close": px[c]})])
|
|
|
|
|
|
# --- 1. rolling windows ------------------------------------------------------------------------
|
|
|
|
def test_rolling_windows_broad_edge_mostly_positive() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_broad_reversion(store)
|
|
rep = walk_forward_oi_deleverage(store, universe=None, cost_bps=5.5, window_days=180)
|
|
store.close()
|
|
rw = rep["rolling_windows"]
|
|
assert rw["n_windows"] >= 3
|
|
assert rw["fraction_positive"] > 0.7
|
|
assert len(rw["windows"]) == rw["n_windows"]
|
|
for w in rw["windows"]:
|
|
assert {"start", "end", "sharpe", "total_return", "days"} <= set(w)
|
|
|
|
|
|
# --- 2. train / holdout DIRECTION test ---------------------------------------------------------
|
|
|
|
def test_train_holdout_direction_generalises_on_broad_edge() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_broad_reversion(store)
|
|
rep = walk_forward_oi_deleverage(store, universe=None, cost_bps=5.5)
|
|
store.close()
|
|
th = rep["train_holdout"]
|
|
assert th["train_direction"] == "reversion"
|
|
assert th["train_sharpe"] > 0.0
|
|
assert th["test_sharpe"] > 0.0
|
|
|
|
|
|
def test_train_holdout_catches_overfit_when_sign_flips_in_test() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_train_only(store)
|
|
rep = walk_forward_oi_deleverage(store, universe=None, cost_bps=5.5)
|
|
store.close()
|
|
th = rep["train_holdout"]
|
|
assert th["train_direction"] == "reversion"
|
|
assert th["train_sharpe"] > 0.0
|
|
assert th["test_sharpe"] < 0.0
|
|
|
|
|
|
# --- 3a. drop-top-N robustness -----------------------------------------------------------------
|
|
|
|
def test_drop_top_n_broad_edge_survives() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_broad_reversion(store)
|
|
rep = walk_forward_oi_deleverage(store, universe=None, cost_bps=5.5)
|
|
store.close()
|
|
dt = rep["drop_top_n"]
|
|
assert dt["full_sharpe"] > 0.0
|
|
assert dt["drop5"]["sharpe"] > 0.0
|
|
assert dt["drop5"]["n_dropped"] >= 1
|
|
assert isinstance(dt["drop5"]["dropped"], list)
|
|
|
|
|
|
def test_drop_top_n_one_coin_edge_collapses() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_one_coin(store)
|
|
rep = walk_forward_oi_deleverage(store, universe=None, cost_bps=5.5)
|
|
store.close()
|
|
dt = rep["drop_top_n"]
|
|
assert dt["full_sharpe"] > 0.0
|
|
assert dt["drop5"]["sharpe"] < dt["full_sharpe"]
|
|
assert dt["drop5"]["sharpe"] < 0.5
|
|
|
|
|
|
# --- 3b. liquidity sweep -----------------------------------------------------------------------
|
|
|
|
def test_liquidity_sweep_returns_metric_per_threshold() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
coins = ("AAAUSDT", "BBBUSDT", "CCCUSDT", "DDDUSDT",
|
|
"EEEUSDT", "FFFUSDT", "GGGUSDT", "HHHUSDT")
|
|
_seed_broad_reversion(store, coins=coins)
|
|
for i in range(722):
|
|
d = _START + i
|
|
for c in coins:
|
|
store.write_features(c, [(d * _DAY, {"turnover": 50_000_000.0})])
|
|
rep = walk_forward_oi_deleverage(store, universe=None, cost_bps=5.5,
|
|
liquidity_grid=(5_000_000.0, 10_000_000.0, 25_000_000.0))
|
|
store.close()
|
|
ls = rep["liquidity_sweep"]
|
|
assert set(ls) == {5_000_000.0, 10_000_000.0, 25_000_000.0}
|
|
for _thr, m in ls.items():
|
|
assert "sharpe" in m and "n_coins" in m
|
|
assert math.isfinite(m["sharpe"])
|
|
assert m["n_coins"] >= 0
|
|
|
|
|
|
# --- 4. look-ahead audit -----------------------------------------------------------------------
|
|
|
|
def test_look_ahead_audit_passes_on_causal_signal() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_broad_reversion(store, days=120)
|
|
rep = look_ahead_audit(store, universe=None)
|
|
store.close()
|
|
assert rep["causal"] is True
|
|
assert rep["leak_days"] == 0
|
|
|
|
|
|
def test_look_ahead_audit_catches_a_deliberately_leaked_mapping() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_broad_reversion(store, days=120)
|
|
rep = look_ahead_audit(store, universe=None, _leak=True) # use the same-day return (leak)
|
|
store.close()
|
|
assert rep["causal"] is False
|
|
assert rep["leak_days"] > 0
|
|
|
|
|
|
def test_walk_forward_includes_look_ahead_audit() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_broad_reversion(store)
|
|
rep = walk_forward_oi_deleverage(store, universe=None, cost_bps=5.5)
|
|
store.close()
|
|
assert rep["look_ahead"]["causal"] is True
|
|
|
|
|
|
# --- 5. PASS/FAIL verdict ----------------------------------------------------------------------
|
|
|
|
def test_verdict_pass_on_broad_generalising_edge() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
coins = ("AAAUSDT", "BBBUSDT", "CCCUSDT", "DDDUSDT",
|
|
"EEEUSDT", "FFFUSDT", "GGGUSDT", "HHHUSDT")
|
|
_seed_broad_reversion(store, coins=coins)
|
|
for i in range(722):
|
|
d = _START + i
|
|
for c in coins:
|
|
store.write_features(c, [(d * _DAY, {"turnover": 50_000_000.0})])
|
|
rep = walk_forward_oi_deleverage(store, universe=None, cost_bps=5.5, window_days=180,
|
|
liquidity_grid=(5_000_000.0, 10_000_000.0, 25_000_000.0))
|
|
store.close()
|
|
v = rep["verdict"]
|
|
assert v["deployable"] is True
|
|
assert v["checks"]["rolling_windows_positive"] is True
|
|
assert v["checks"]["oos_direction_positive"] is True
|
|
assert v["checks"]["survives_drop_top5"] is True
|
|
assert v["checks"]["liquidity_robust"] is True
|
|
assert v["checks"]["look_ahead_clean"] is True
|
|
|
|
|
|
def test_verdict_fail_on_overfit_edge() -> None:
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_train_only(store)
|
|
rep = walk_forward_oi_deleverage(store, universe=None, cost_bps=5.5)
|
|
store.close()
|
|
v = rep["verdict"]
|
|
assert v["deployable"] is False
|
|
assert v["checks"]["oos_direction_positive"] is False
|
|
|
|
|
|
# --- 6. read-only ------------------------------------------------------------------------------
|
|
|
|
def test_walk_forward_is_read_only() -> None:
|
|
inner = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_broad_reversion(inner)
|
|
store = _WriteTripwireStore(inner)
|
|
rep = walk_forward_oi_deleverage(store, universe=None, cost_bps=5.5)
|
|
inner.close()
|
|
assert rep["rolling_windows"]["n_windows"] >= 1
|
|
|
|
|
|
# --- 7. CLI ------------------------------------------------------------------------------------
|
|
|
|
def test_cli_walk_forward_prints_verdict(monkeypatch) -> None:
|
|
from typer.testing import CliRunner
|
|
|
|
import fxhnt.cli as cli
|
|
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_broad_reversion(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", "--walk-forward"])
|
|
store.close()
|
|
assert result.exit_code == 0, result.output
|
|
out = result.output.lower()
|
|
assert "walk-forward" in out or "walk forward" in out
|
|
assert "rolling" in out
|
|
assert "oos" in out or "holdout" in out
|
|
assert "verdict" in out
|
|
assert "pass" in out or "fail" in out
|