240 lines
11 KiB
Python
240 lines
11 KiB
Python
"""OOS test of a PURE drawdown-kill on the combined Bybit 4-edge book — READ-ONLY.
|
|
|
|
The combined Bybit book (naive eq-wt of tstrend+unlock+carry+positioning) is the deployable book (OOS Sharpe
|
|
~2.43, maxDD ~-13%). The full risk OVERLAY was rejected — it over-levered and lost OOS. A PURE drawdown-kill
|
|
is a DIFFERENT, simpler thing: capital preservation only. It flattens to cash on a tail drawdown (exposure 0)
|
|
and re-enters per the existing `DrawdownKillSwitch` hysteresis (exposure 1) — NO leverage change EVER. The
|
|
overlay A/B's kill rode on TOP of vol-target leverage; this is the kill ALONE.
|
|
|
|
These tests prove:
|
|
* the naive eq-wt combined book is built from `bybit_sleeve_ret` (repo or prebuilt dict);
|
|
* the pure kill flattens (return 0) while in the drawdown-kill state, re-enters per the logic; exposure is
|
|
0/1 only — NEVER levered;
|
|
* a sharp-tail-drawdown fixture → the kill REDUCES maxDD (DD benefit shown);
|
|
* a choppy fixture → the kill WHIPSAWS (Sharpe loss without DD benefit, flagged);
|
|
* train/test verdict: a config that helps on TRAIN but not TEST is NOT adopted (OOS discipline); naive is
|
|
chosen when no kill clears the DD+Sharpe bar;
|
|
* READ-ONLY (WriteTripwire).
|
|
NO network — the repo is in-memory; external inputs are injected.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
|
|
from fxhnt.application.bybit_drawdown_kill_test import (
|
|
KILL_CONFIG_NAMES,
|
|
_pure_kill_curve,
|
|
drawdown_kill_test,
|
|
)
|
|
|
|
|
|
# --- fixtures ----------------------------------------------------------------------------------
|
|
|
|
def _flat_up_then_crash(start: int = 0, up_days: int = 120, crash_days: int = 30,
|
|
up: float = 0.004, crash: float = -0.02) -> dict[int, float]:
|
|
"""A book that rises steadily then suffers a sharp sustained tail drawdown — the case a kill is FOR."""
|
|
series: dict[int, float] = {}
|
|
d = start
|
|
for _ in range(up_days):
|
|
series[d] = up
|
|
d += 1
|
|
for _ in range(crash_days):
|
|
series[d] = crash
|
|
d += 1
|
|
return series
|
|
|
|
|
|
def _choppy(days: int = 200, amp: float = 0.03) -> dict[int, float]:
|
|
"""A choppy oscillating book with no sustained drawdown — the case a kill WHIPSAWS on."""
|
|
return {d: amp * math.sin(0.7 * d) for d in range(days)}
|
|
|
|
|
|
def _two_sleeves_from(book: dict[int, float]) -> dict[str, dict[int, float]]:
|
|
"""Make an eq-wt of two identical sleeves reproduce `book` exactly (eq-wt of x,x = x)."""
|
|
return {"crypto_tstrend": dict(book), "xsfunding": dict(book)}
|
|
|
|
|
|
class _FakeRepo:
|
|
"""A stand-in for PaperRepo exposing only `bybit_sleeve_returns()` (the precomputed table reader)."""
|
|
|
|
def __init__(self, returns_by_sleeve: dict[str, dict[int, float]]) -> None:
|
|
self._r = {s: dict(v) for s, v in returns_by_sleeve.items()}
|
|
|
|
def bybit_sleeve_returns(self) -> dict[str, dict[int, float]]:
|
|
return {s: dict(v) for s, v in self._r.items()}
|
|
|
|
|
|
class _WriteTripwireRepo:
|
|
_FORBIDDEN = frozenset({
|
|
"upsert_bybit_sleeve_ret", "replace_positions", "upsert_nav", "upsert_sleeve_ret",
|
|
"replace_shadow_positions", "replace_trades", "_create_schema",
|
|
})
|
|
|
|
def __init__(self, inner: _FakeRepo) -> None:
|
|
object.__setattr__(self, "_inner", inner)
|
|
|
|
def __getattr__(self, name: str):
|
|
if name in _WriteTripwireRepo._FORBIDDEN:
|
|
raise AssertionError(f"READ-ONLY violation: drawdown-kill test called write method {name!r}")
|
|
return getattr(self._inner, name)
|
|
|
|
|
|
# --- the pure kill: flatten / re-enter / exposure is 0|1 only ----------------------------------
|
|
|
|
def test_pure_kill_curve_exposure_is_zero_or_one_never_levered() -> None:
|
|
"""The pure-kill curve return on each day equals either the book return (exposure 1) or 0 (exposure 0,
|
|
flattened) — it is NEVER a scaled / levered multiple of the book return."""
|
|
book = _flat_up_then_crash()
|
|
killed, exposure = _pure_kill_curve(book, kill_dd=0.10)
|
|
for d, r in killed.items():
|
|
assert exposure[d] in (0.0, 1.0), (d, exposure[d])
|
|
expected = exposure[d] * book[d]
|
|
assert abs(r - expected) < 1e-12, (d, r, expected)
|
|
|
|
|
|
def test_pure_kill_flattens_during_drawdown_state() -> None:
|
|
"""Once the running drawdown breaches the kill threshold, the pure kill flattens (return 0) on subsequent
|
|
days until re-entry — so at least one crash day is zeroed out."""
|
|
book = _flat_up_then_crash(up=0.004, crash=-0.03, crash_days=40)
|
|
killed, exposure = _pure_kill_curve(book, kill_dd=0.10)
|
|
flattened = [d for d, e in exposure.items() if e == 0.0]
|
|
assert flattened, "expected the kill to flatten at least one day during the crash"
|
|
# every flattened day contributes exactly 0 to the curve.
|
|
for d in flattened:
|
|
assert killed[d] == 0.0
|
|
|
|
|
|
def test_pure_kill_reenters_after_recovery() -> None:
|
|
"""After a kill, when the shadow drawdown recovers below the re-enter threshold the exposure returns to 1
|
|
(the book is re-entered) — so a kill is NOT permanent."""
|
|
# crash then a long recovery rally.
|
|
book = _flat_up_then_crash(up_days=80, crash_days=20, up=0.004, crash=-0.03)
|
|
d = max(book) + 1
|
|
for _ in range(200):
|
|
book[d] = 0.01 # strong recovery
|
|
d += 1
|
|
_killed, exposure = _pure_kill_curve(book, kill_dd=0.10)
|
|
got_killed = any(e == 0.0 for e in exposure.values())
|
|
reentered = exposure[max(exposure)] == 1.0
|
|
assert got_killed and reentered, (got_killed, reentered)
|
|
|
|
|
|
# --- DD benefit on a tail drawdown; whipsaw on a choppy book -----------------------------------
|
|
|
|
def test_kill_reduces_maxdd_on_sharp_tail_drawdown() -> None:
|
|
"""On a book with a sharp sustained tail drawdown, a pure kill REDUCES maxDD vs naive (the DD benefit)."""
|
|
book = _flat_up_then_crash(up=0.004, crash=-0.025, crash_days=40)
|
|
rep = drawdown_kill_test(_two_sleeves_from(book), train_frac=0.6)
|
|
assert rep["available"] is True
|
|
naive_dd = rep["table"]["naive"]["full"]["max_dd"]
|
|
# the deepest-threshold kill that actually triggers should cut maxDD.
|
|
best_dd = min(rep["table"][c]["full"]["max_dd"] for c in KILL_CONFIG_NAMES)
|
|
# max_dd is <= 0; "less deep" = closer to 0 = larger value.
|
|
assert best_dd > naive_dd + 1e-9, (best_dd, naive_dd)
|
|
# and at least one kill config actually triggered.
|
|
assert any(rep["table"][c]["full"]["triggers"] > 0 for c in KILL_CONFIG_NAMES)
|
|
|
|
|
|
def test_kill_whipsaws_on_choppy_book() -> None:
|
|
"""On a choppy book with no sustained drawdown, a tight kill that triggers should COST Sharpe without a
|
|
meaningful DD benefit — the whipsaw case. The verdict must NOT adopt a whipsawing kill."""
|
|
book = _choppy(days=240, amp=0.03)
|
|
rep = drawdown_kill_test(_two_sleeves_from(book), train_frac=0.6)
|
|
assert rep["available"] is True
|
|
# naive is the recommendation when no kill clears the DD+Sharpe bar on choppy data.
|
|
assert rep["verdict"]["recommendation"] == "naive", rep["verdict"]
|
|
|
|
|
|
# --- train/test OOS discipline -----------------------------------------------------------------
|
|
|
|
def test_kill_helps_train_not_test_is_not_adopted() -> None:
|
|
"""A kill that cuts DD on TRAIN but whipsaws on TEST (no OOS DD benefit) must NOT be adopted — the
|
|
recommendation falls back to naive. Fixture: a tail crash early (TRAIN) then choppy (TEST)."""
|
|
book = _flat_up_then_crash(up_days=80, crash_days=40, up=0.004, crash=-0.025)
|
|
d = max(book) + 1
|
|
# TEST region: choppy, no sustained drawdown — a kill only whipsaws here.
|
|
for k in range(160):
|
|
book[d] = 0.025 * math.sin(0.7 * k)
|
|
d += 1
|
|
rep = drawdown_kill_test(_two_sleeves_from(book), train_frac=0.5)
|
|
assert rep["available"] is True
|
|
rec = rep["verdict"]["recommendation"]
|
|
# whatever is recommended must satisfy the OOS bar (or be naive); a TRAIN-only winner is not adopted.
|
|
if rec != "naive":
|
|
te = rep["table"][rec]["test"]
|
|
naive_te = rep["table"]["naive"]["test"]
|
|
rel_dd_red = (naive_te["max_dd"] - te["max_dd"]) / abs(naive_te["max_dd"]) if naive_te["max_dd"] < 0 else 0.0
|
|
assert rel_dd_red >= rep["verdict"]["dd_bar"] - 1e-9
|
|
assert (naive_te["sharpe"] - te["sharpe"]) <= rep["verdict"]["sharpe_bar"] + 1e-9
|
|
|
|
|
|
def test_naive_chosen_when_no_kill_clears_the_bar() -> None:
|
|
"""When the book's drawdown is already contained (no tail event), every kill barely triggers → no DD
|
|
benefit → STAY NAIVE."""
|
|
book = {d: 0.003 + 0.005 * math.sin(0.3 * d) for d in range(300)} # mild, well-contained
|
|
rep = drawdown_kill_test(_two_sleeves_from(book), train_frac=0.6)
|
|
assert rep["available"] is True
|
|
assert rep["verdict"]["recommendation"] == "naive"
|
|
|
|
|
|
# --- repo path + read-only ---------------------------------------------------------------------
|
|
|
|
def test_reads_combined_book_from_repo() -> None:
|
|
"""drawdown_kill_test accepts a repo and builds the naive eq-wt combined book from bybit_sleeve_returns()."""
|
|
book = _flat_up_then_crash(up=0.004, crash=-0.025, crash_days=30)
|
|
repo = _FakeRepo(_two_sleeves_from(book))
|
|
rep = drawdown_kill_test(repo, train_frac=0.6)
|
|
assert rep["available"] is True
|
|
assert set(rep["table"]) == {"naive", *KILL_CONFIG_NAMES}
|
|
|
|
|
|
def test_drawdown_kill_test_is_read_only() -> None:
|
|
book = _flat_up_then_crash()
|
|
guarded = _WriteTripwireRepo(_FakeRepo(_two_sleeves_from(book)))
|
|
rep = drawdown_kill_test(guarded, train_frac=0.6)
|
|
assert rep["available"] is True
|
|
|
|
|
|
def test_empty_repo_reports_unavailable() -> None:
|
|
rep = drawdown_kill_test(_FakeRepo({}), train_frac=0.6)
|
|
assert rep["available"] is False
|
|
assert "reason" in rep
|
|
|
|
|
|
# --- table shape: train + test + full per config, with triggers --------------------------------
|
|
|
|
def test_table_reports_train_test_full_with_triggers() -> None:
|
|
book = _flat_up_then_crash(up=0.004, crash=-0.02, crash_days=40)
|
|
rep = drawdown_kill_test(_two_sleeves_from(book), train_frac=0.6)
|
|
assert rep["available"] is True
|
|
for name, row in rep["table"].items():
|
|
for split in ("train", "test", "full"):
|
|
m = row[split]
|
|
assert math.isfinite(m["sharpe"]), (name, split)
|
|
assert math.isfinite(m["cagr"]), (name, split)
|
|
assert m["max_dd"] <= 1e-9, (name, split)
|
|
assert m["days"] >= 0
|
|
assert m["triggers"] >= 0
|
|
|
|
|
|
# --- CLI smoke (mocked repo) -------------------------------------------------------------------
|
|
|
|
def test_cli_bybit_drawdown_kill_test_prints_table(monkeypatch) -> None:
|
|
from typer.testing import CliRunner
|
|
|
|
import fxhnt.cli as cli
|
|
|
|
book = _flat_up_then_crash(up=0.004, crash=-0.025, crash_days=40)
|
|
repo = _FakeRepo(_two_sleeves_from(book))
|
|
|
|
monkeypatch.setattr(
|
|
"fxhnt.adapters.persistence.paper_repo.PaperRepo", lambda *a, **k: repo, raising=False)
|
|
|
|
result = CliRunner().invoke(cli.app, ["bybit-drawdown-kill-test", "--train-frac", "0.6"])
|
|
assert result.exit_code == 0, result.output
|
|
out_lower = result.output.lower()
|
|
assert "drawdown" in out_lower and "kill" in out_lower
|
|
assert "naive" in out_lower
|
|
assert "train" in out_lower and "test" in out_lower
|
|
assert "recommend" in out_lower or "verdict" in out_lower
|