346 lines
17 KiB
Python
346 lines
17 KiB
Python
"""STEP 3 (final) of the "spread-as-a-strat" research — the SPREAD-AWARE EXECUTION OVERLAY.
|
||
|
||
Steps 1-2 falsified both DIRECTIONAL spread ideas (static illiquidity premium = cost trap on entry/holding;
|
||
deterioration shorts = cost trap on the widened exit). Conclusion: the spread is a COST signal, not alpha.
|
||
Step 3 tests the NON-DIRECTIONAL use — use the measured spread to AVOID cost on the edges we already run — as
|
||
a SHADOW eval that never changes the live book's returns.
|
||
|
||
Two overlay mechanisms vs the baseline, NET of the MEASURED per-coin cost (Bybit 5.5bp taker + half the
|
||
Corwin–Schultz spread):
|
||
(A) BASELINE — the current naive eq-wt book (k=0, λ=0).
|
||
(B) SPREAD-AWARE SIZING — scale weight by 1/(1+k·spread_bps) then renormalise (gross preserved).
|
||
(C) NO-TRADE BANDS — only rebalance coin i when |Δw_i| exceeds a cost-proportional band λ·cost_i/1e4.
|
||
(B+C) combined.
|
||
|
||
Tests (TDD):
|
||
* sizing downweights high-spread coins AND renormalises (Σ|w| == the gross target); no-lookahead; sign kept;
|
||
* sizing is a NO-OP when k=0 or all spreads are 0;
|
||
* the no-trade band SKIPS sub-threshold rebalances → lower turnover than baseline; a BIG signal change still
|
||
trades;
|
||
* on a fixture with COSTLY high-spread names, the overlay's NET > baseline NET (a real lift);
|
||
* on an ALL-CHEAP fixture, the lift ≈ 0 (the overlay is a no-op when nothing is costly — proves it is not a
|
||
free lunch);
|
||
* baseline (k=0, λ=0) reconciles to the naive per-coin book (no engine drift);
|
||
* the eval/sweep run on a seeded store with the OOS split + verdict; the breadth table covers the universes;
|
||
* READ-ONLY (WriteTripwire) + determinism + CLI smoke.
|
||
NO network — the store is in-memory SQLite; external inputs are injected.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
|
||
from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
|
||
from fxhnt.application.bybit_liquidity_sweep import per_coin_book_returns
|
||
from fxhnt.application.spread_overlay import (
|
||
_overlay_series,
|
||
evaluate_overlay,
|
||
per_coin_weight_panel,
|
||
sized_weights,
|
||
spread_overlay_eval,
|
||
spread_overlay_test,
|
||
)
|
||
|
||
_DAY = 86_400
|
||
|
||
|
||
# --- fixtures (mirror the liquidity-sweep fixtures) ----------------------------------------------
|
||
|
||
def _ohlc_from_close(close: float, spread_frac: float, vol_frac: float, phase: float) -> tuple[float, float]:
|
||
vol = vol_frac * (0.5 + 0.5 * math.cos(7.0 * phase))
|
||
half = 0.5 * spread_frac + vol
|
||
return close * (1.0 + half), close * (1.0 - half)
|
||
|
||
|
||
def _seed_carry(store: TimescaleFeatureStore, funding: dict[str, float], spread_by_sym: dict[str, float], *,
|
||
days: int = 400) -> list[str]:
|
||
"""A carry (xsfunding) book: half positive-funding, half negative-funding coins, each with a KNOWN
|
||
injected proportional spread so its MEASURED cost is the controlled driver."""
|
||
for idx, (sym, fund) in enumerate(funding.items()):
|
||
sf = spread_by_sym[sym]
|
||
rows = []
|
||
for d in range(days):
|
||
f = fund + 0.0003 * math.cos(0.5 * d + idx)
|
||
px = 100.0
|
||
high, low = _ohlc_from_close(px, sf, vol_frac=0.0005, phase=0.5 * d + idx)
|
||
rows.append((d * _DAY, {"funding": f, "close": px, "spot_close": px, "high": high, "low": low}))
|
||
store.write_features(sym, rows)
|
||
return list(funding)
|
||
|
||
|
||
def _seed_turnover(store: TimescaleFeatureStore, dv: dict[str, float], *, days: int = 400) -> None:
|
||
for sym, v in dv.items():
|
||
store.write_features(sym, [(d * _DAY, {"turnover": v}) for d in range(days)])
|
||
|
||
|
||
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: spread-overlay called write method {name!r}")
|
||
return getattr(self._inner, name)
|
||
|
||
|
||
# --- (B) spread-aware sizing (pure) --------------------------------------------------------------
|
||
|
||
def test_sizing_downweights_high_spread_and_preserves_gross() -> None:
|
||
"""Sizing scales by 1/(1+k·spread): the high-spread coin is downweighted relative to the low-spread coin,
|
||
and the renormalisation keeps the GROSS exposure Σ|w| equal to the original target (budget redistributed,
|
||
not reduced). Signs are preserved."""
|
||
weights = {"CHEAP": 0.5, "EXP": 0.5}
|
||
spread = {"CHEAP": 0.0, "EXP": 1000.0}
|
||
sized = sized_weights(weights, spread, k=0.01)
|
||
assert sized["CHEAP"] > sized["EXP"] # expensive coin downweighted
|
||
assert sized["CHEAP"] > weights["CHEAP"] # budget redistributed onto the cheap coin
|
||
assert abs(sum(abs(w) for w in sized.values()) - 1.0) < 1e-12 # gross exposure preserved
|
||
assert all((sized[c] > 0) == (weights[c] > 0) for c in weights) # signs kept
|
||
|
||
|
||
def test_sizing_preserves_gross_with_shorts() -> None:
|
||
"""Gross exposure (Σ|w|) is preserved even with a mix of long and short weights."""
|
||
weights = {"A": 0.6, "B": -0.4, "C": 0.2}
|
||
spread = {"A": 5.0, "B": 300.0, "C": 50.0}
|
||
sized = sized_weights(weights, spread, k=0.05)
|
||
assert abs(sum(abs(w) for w in sized.values()) - sum(abs(w) for w in weights.values())) < 1e-12
|
||
assert sized["B"] < 0 # short stays short
|
||
|
||
|
||
def test_sizing_is_noop_when_k_zero_or_all_cheap() -> None:
|
||
weights = {"A": 0.5, "B": -0.5}
|
||
assert sized_weights(weights, {"A": 100.0, "B": 200.0}, k=0.0) == weights
|
||
# all spreads zero → factor 1 everywhere → unchanged (within fp).
|
||
sized = sized_weights(weights, {"A": 0.0, "B": 0.0}, k=0.5)
|
||
assert all(abs(sized[c] - weights[c]) < 1e-12 for c in weights)
|
||
|
||
|
||
# --- (C) the no-trade band -----------------------------------------------------------------------
|
||
|
||
def _wiggle_panel(n_days: int = 60) -> dict[int, dict[str, tuple[float, float]]]:
|
||
"""A panel whose target weight WIGGLES by a tiny amount each day (sub-band churn) — the band should skip
|
||
most of these rebalances. Per-unit return ~0 so the test isolates turnover."""
|
||
daily: dict[int, dict[str, tuple[float, float]]] = {}
|
||
for d in range(n_days):
|
||
w = 0.50 + 0.002 * math.sin(0.7 * d) # tiny daily churn around 0.5
|
||
daily[d] = {"X": (w, 0.0)}
|
||
return daily
|
||
|
||
|
||
def test_no_trade_band_reduces_turnover() -> None:
|
||
daily = _wiggle_panel()
|
||
cost = {"X": 50.0} # 50bp cost → band at λ=4 is 0.02 (>> the 0.002 wiggle) so churn is skipped
|
||
spread = {"X": 100.0}
|
||
_g0, _n0, turn_base, _c0 = _overlay_series(daily, cost, spread, k=0.0, lam=0.0)
|
||
_g1, _n1, turn_band, _c1 = _overlay_series(daily, cost, spread, k=0.0, lam=4.0)
|
||
assert turn_band < turn_base, (turn_base, turn_band)
|
||
|
||
|
||
def test_no_trade_band_still_trades_a_big_signal_change() -> None:
|
||
"""A wiggle panel with one BIG jump: the band skips the small churn but the large move clears the band and
|
||
DOES trade (the held weight moves to the new target)."""
|
||
daily = _wiggle_panel(n_days=30)
|
||
daily[15] = {"X": (0.95, 0.0)} # a big jump well above any cost-proportional band
|
||
cost = {"X": 50.0}
|
||
spread = {"X": 100.0}
|
||
_gross, _net, turn, _cd = _overlay_series(daily, cost, spread, k=0.0, lam=4.0)
|
||
# turnover is positive (it traded the big move) but far below the always-trade baseline.
|
||
_g, _n, turn_base, _c = _overlay_series(daily, cost, spread, k=0.0, lam=0.0)
|
||
assert 0.0 < turn < turn_base
|
||
# the held weight after the jump must have moved to ~0.95 (it traded the big change): the very next day's
|
||
# small wiggle back toward 0.5 is within the band, so it should still be holding ~0.95 → big |Δ| trade
|
||
# already happened. Re-derive via a 2-day mini panel to assert the trade fired on the jump day.
|
||
mini = {0: {"X": (0.5, 0.0)}, 1: {"X": (0.95, 0.0)}}
|
||
_gg, _nn, mini_turn, _cc = _overlay_series(mini, cost, spread, k=0.0, lam=4.0)
|
||
assert mini_turn >= 0.45 - 1e-9 # 0.5 open + ~0.45 jump traded
|
||
|
||
|
||
# --- baseline reconciliation ---------------------------------------------------------------------
|
||
|
||
def test_baseline_reconciles_to_naive_per_coin_book() -> None:
|
||
"""k=0, λ=0 must reproduce the naive eq-wt per-coin book EXACTLY (no engine drift) — same gross and net."""
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
funding = {"AAAUSDT": 0.001, "BBBUSDT": 0.001, "XXXUSDT": -0.001, "YYYUSDT": -0.001}
|
||
spread = {s: 0.0008 for s in funding}
|
||
_seed_carry(store, funding, spread)
|
||
_seed_turnover(store, {s: 50_000_000.0 for s in funding})
|
||
daily, cost, spr = per_coin_weight_panel(store, universe=None, sleeves=["xsfunding"], unlock_events=[])
|
||
g, n, _t, _cd = _overlay_series(daily, cost, spr, k=0.0, lam=0.0)
|
||
pc = per_coin_book_returns(store, universe=None, sleeves=["xsfunding"], unlock_events=[])
|
||
store.close()
|
||
common = sorted(set(g) & set(pc["gross_series"]))
|
||
assert common
|
||
for d in common:
|
||
assert abs(g[d] - pc["gross_series"][d]) < 1e-12, (d, g[d], pc["gross_series"][d])
|
||
assert abs(n[d] - pc["net_series"][d]) < 1e-12, (d, n[d], pc["net_series"][d])
|
||
|
||
|
||
# --- the lift + no-free-lunch tests (the heart of Step 3) ----------------------------------------
|
||
|
||
def _costly_panel(n_days: int = 80) -> dict[int, dict[str, tuple[float, float]]]:
|
||
"""A 2-coin book where a CHEAP coin carries the alpha (steady positive contribution, near-zero cost) and an
|
||
EXPENSIVE coin is dead weight: zero return but it FLIPS sign every day, so the baseline pays its wide
|
||
spread on a full turnover EVERY day. The spread-aware overlay should downweight / stop-churning the
|
||
expensive coin and lift NET."""
|
||
daily: dict[int, dict[str, tuple[float, float]]] = {}
|
||
for d in range(n_days):
|
||
cheap_w = 0.5
|
||
cheap_ret = 0.0015 # steady positive per-unit return
|
||
exp_w = 0.5 if d % 2 == 0 else -0.5 # flips every day → full turnover each day
|
||
daily[d] = {"CHEAP": (cheap_w, cheap_w * cheap_ret), "EXP": (exp_w, 0.0)}
|
||
return daily
|
||
|
||
|
||
def test_overlay_lifts_net_on_costly_fixture() -> None:
|
||
"""COSTLY fixture: the best overlay's NET-FULL Sharpe beats the baseline (the spread is used to AVOID the
|
||
expensive coin's cost — the whole point of Step 3)."""
|
||
daily = _costly_panel()
|
||
cost = {"CHEAP": 6.0, "EXP": 300.0} # expensive coin: wide measured spread → big cost
|
||
spread = {"CHEAP": 1.0, "EXP": 600.0}
|
||
base = evaluate_overlay(daily, cost, spread, k=0.0, lam=0.0)
|
||
sizing = evaluate_overlay(daily, cost, spread, k=0.05, lam=0.0)
|
||
band = evaluate_overlay(daily, cost, spread, k=0.0, lam=4.0)
|
||
best = max((sizing, band), key=lambda r: r["net_full_sharpe"])
|
||
assert best["net_full_sharpe"] > base["net_full_sharpe"], (base["net_full_sharpe"],
|
||
best["net_full_sharpe"])
|
||
# and the cost drag / turnover are reduced.
|
||
assert best["cost_drag_bps"] <= base["cost_drag_bps"] + 1e-9
|
||
assert best["turnover"] <= base["turnover"] + 1e-9
|
||
|
||
|
||
def _all_cheap_panel(n_days: int = 80) -> dict[int, dict[str, tuple[float, float]]]:
|
||
"""Same SHAPE as the costly panel but every coin is DEEP (≈ zero spread / fee-floor cost). The overlay
|
||
should be ≈ a no-op — there is no cost to avoid (proves it is not a free lunch)."""
|
||
daily: dict[int, dict[str, tuple[float, float]]] = {}
|
||
for d in range(n_days):
|
||
a_w = 0.5
|
||
a_ret = 0.0015
|
||
b_w = 0.5 if d % 2 == 0 else -0.5
|
||
daily[d] = {"A": (a_w, a_w * a_ret), "B": (b_w, 0.0)}
|
||
return daily
|
||
|
||
|
||
def test_overlay_is_noop_when_nothing_is_costly() -> None:
|
||
"""ALL-CHEAP fixture: spreads ≈ 0 → sizing is a no-op (identical net series); the lift is ≈ 0. The overlay
|
||
only helps when there is genuine cost to avoid."""
|
||
daily = _all_cheap_panel()
|
||
cost = {"A": 5.5, "B": 5.5} # fee floor only
|
||
spread = {"A": 0.0, "B": 0.0}
|
||
base = evaluate_overlay(daily, cost, spread, k=0.0, lam=0.0)
|
||
sizing = evaluate_overlay(daily, cost, spread, k=0.2, lam=0.0)
|
||
# sizing with zero spreads is exactly the baseline.
|
||
assert sizing["net_series"] == base["net_series"]
|
||
# the band may trim a hair of turnover, but the Sharpe lift is negligible (no costly coin to avoid).
|
||
band = evaluate_overlay(daily, cost, spread, k=0.0, lam=4.0)
|
||
assert band["net_full_sharpe"] - base["net_full_sharpe"] < 0.10
|
||
|
||
|
||
# --- the one-universe eval + the breadth sweep ---------------------------------------------------
|
||
|
||
def _seed_two_tier(store: TimescaleFeatureStore, *, days: int = 400) -> None:
|
||
"""A liquid tier (tight spread, high $vol) + a thin tier (WIDE spread, low $vol) carry book — the broad
|
||
universe pulls in the thin coins the overlay can help with; the $10M bound excludes them."""
|
||
deep = {"AAAUSDT": 0.001, "BBBUSDT": 0.001, "CCCUSDT": 0.001,
|
||
"XXXUSDT": -0.001, "YYYUSDT": -0.001, "ZZZUSDT": -0.001}
|
||
thin = {"PPPUSDT": 0.0012, "QQQUSDT": 0.0012, "RRRUSDT": 0.0012,
|
||
"SSSUSDT": -0.0012, "TTTUSDT": -0.0012, "UUUUSDT": -0.0012}
|
||
spread = {s: 0.0004 for s in deep}
|
||
spread.update({s: 0.02 for s in thin}) # 200bp measured spread on the thin tier
|
||
_seed_carry(store, {**deep, **thin}, spread, days=days)
|
||
dv = {s: 100_000_000.0 for s in deep}
|
||
dv.update({s: 2_000_000.0 for s in thin}) # thin coins clear $1M but fail the $10M bound
|
||
_seed_turnover(store, dv, days=days)
|
||
|
||
|
||
def test_eval_runs_with_oos_split_and_verdict() -> None:
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_two_tier(store)
|
||
rep = spread_overlay_eval(store, universe=None, sleeves=["xsfunding"], unlock_events=[])
|
||
store.close()
|
||
assert rep["available"] is True
|
||
for variant in ("baseline", "sizing", "band", "combined"):
|
||
r = rep[variant]
|
||
assert math.isfinite(r["net_test_sharpe"])
|
||
assert math.isfinite(r["cost_drag_bps"])
|
||
assert math.isfinite(r["turnover"])
|
||
assert rep["verdict"]["verdict"] in ("CONFIRM", "MARGINAL", "NO-HELP")
|
||
assert rep["verdict"]["best_variant"] in ("sizing", "band", "combined")
|
||
|
||
|
||
def test_breadth_sweep_covers_universes_and_lift_grows_with_breadth() -> None:
|
||
"""The broad universe (which includes the thin, wide-spread coins) should see at least as large an overlay
|
||
cost benefit as the liquid $10M bound — the overlay's value scales with breadth."""
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_two_tier(store)
|
||
rep = spread_overlay_test(store, min_dollar_vols=(10_000_000.0, 0.0),
|
||
sleeves=["xsfunding"], unlock_events=[])
|
||
store.close()
|
||
assert rep["available"] is True
|
||
by_label = {row["universe_label"]: row for row in rep["table"]}
|
||
assert "broad/all" in by_label
|
||
broad = by_label["broad/all"]
|
||
# the broad universe carries the wide-spread thin coins → a larger median measured spread than the bounded
|
||
# book, so there is genuinely more cost for the overlay to avoid.
|
||
if "$10M" in by_label:
|
||
assert broad["median_spread_bps"] >= by_label["$10M"]["median_spread_bps"] - 1e-9
|
||
# the overlay finds a cost-drag reduction on the broad book.
|
||
assert broad["verdict"]["d_cost_drag_bps"] >= -1e-9 or broad["verdict"]["d_sharpe"] >= 0.0
|
||
|
||
|
||
# --- READ-ONLY + determinism ---------------------------------------------------------------------
|
||
|
||
def test_eval_is_read_only() -> None:
|
||
inner = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_two_tier(inner)
|
||
guarded = _WriteTripwireStore(inner)
|
||
rep = spread_overlay_eval(guarded, universe=None, sleeves=["xsfunding"], unlock_events=[])
|
||
inner.close()
|
||
assert rep["available"] is True
|
||
|
||
|
||
def test_eval_is_deterministic() -> None:
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_two_tier(store)
|
||
a = spread_overlay_eval(store, universe=None, sleeves=["xsfunding"], unlock_events=[])
|
||
b = spread_overlay_eval(store, universe=None, sleeves=["xsfunding"], unlock_events=[])
|
||
store.close()
|
||
assert a["baseline"]["net_series"] == b["baseline"]["net_series"]
|
||
assert a["combined"]["net_series"] == b["combined"]["net_series"]
|
||
|
||
|
||
def test_empty_store_reports_unavailable() -> None:
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
rep = spread_overlay_eval(store, universe=None, sleeves=["xsfunding"], unlock_events=[])
|
||
store.close()
|
||
assert rep["available"] is False
|
||
|
||
|
||
# --- CLI smoke (mocked store) --------------------------------------------------------------------
|
||
|
||
def test_cli_spread_overlay_test_prints_table(monkeypatch) -> None:
|
||
from typer.testing import CliRunner
|
||
|
||
import fxhnt.cli as cli
|
||
|
||
seeded = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_two_tier(seeded, days=400)
|
||
|
||
monkeypatch.setattr(
|
||
"fxhnt.adapters.warehouse.timescale_feature_store.TimescaleFeatureStore",
|
||
lambda *a, **k: seeded, raising=False)
|
||
|
||
result = CliRunner().invoke(
|
||
cli.app, ["spread-overlay-test", "--min-dollar-vols", "10000000,0", "--sleeves", "xsfunding"])
|
||
seeded.close()
|
||
|
||
assert result.exit_code == 0, result.output
|
||
out = result.output.lower()
|
||
assert "universe" in out
|
||
assert "baseline" in out or "net" in out
|
||
assert "verdict" in out
|