Jeroen Grusewski
8bbe7fb123
Add POSITIONING_COIN_GROSS_CAP constant (0.07) to bybit_forward_track.py with full validation comment. Wire the constant through bybit_book_persist.py's sleeve-return computation. Add test verifying the constant value. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
466 lines
22 KiB
Python
466 lines
22 KiB
Python
"""READ-ONLY evaluator for the Bybit POSITIONING edge (long/short account-ratio contrarian / momentum).
|
||
|
||
The signal is the Bybit long/short ACCOUNT ratio (retail positioning). The CONTRARIAN edge SHORTS coins
|
||
where retail is over-LONG (high buyRatio) and LONGS coins where retail is over-SHORT (low buyRatio), betting
|
||
on PRICE (close-to-close) reversion — cross-sectional, market-neutral. MOMENTUM is the exact sign-flip.
|
||
|
||
Tests assert:
|
||
* contrarian weights short high-ratio / long low-ratio, market-neutral (Σw≈0) + unit-gross (Σ|w|≈1);
|
||
* momentum = exact negation of contrarian;
|
||
* on a fixture where fading retail is profitable, the contrarian 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 the
|
||
correlation fields vs ALL THREE existing edges (tstrend, unlock, xsfunding);
|
||
* 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_positioning_eval import (
|
||
positioning_metrics_from_store,
|
||
positioning_weights,
|
||
verify_positioning_edge,
|
||
)
|
||
|
||
_DAY = 86_400
|
||
|
||
|
||
# --- capturability haircuts (capacity + tail-concentration) -------------------------------------
|
||
|
||
def test_robust_sigma_is_mad_based_and_spike_immune() -> None:
|
||
import statistics as _st
|
||
|
||
from fxhnt.application.bybit_positioning_eval import _robust_sigma
|
||
calm = [0.01, -0.01, 0.008, -0.012, 0.005, -0.007, 0.009, -0.006]
|
||
base = _robust_sigma(calm)
|
||
assert base > 0.0
|
||
# one huge spike: a plain std explodes, but the MAD-based robust σ barely moves (spike-immune — the whole
|
||
# point, so the tail-cap's own reference is not inflated by the days it exists to clip).
|
||
with_spike = _robust_sigma([*calm, 1.0])
|
||
assert with_spike < 0.25 * _st.pstdev([*calm, 1.0])
|
||
assert _robust_sigma([]) == 0.0
|
||
|
||
|
||
def test_capacity_haircut_caps_gain_only_at_scale() -> None:
|
||
from fxhnt.application.bybit_positioning_eval import positioning_returns_from_store
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
# 2 coins, STABLE contrarian weights: AAA over-long (short, w<0), BBB over-short (long, w>0). BBB pumps
|
||
# +100% on the last transition; BBB turnover is modest so a LARGE book cannot exit that gain at size.
|
||
bbb_close = [100.0, 100.0, 100.0, 200.0] # +100% booked on epoch-day 2 (its next-day return)
|
||
for d in range(4):
|
||
store.write_features("AAAUSDT", [(d * _DAY, {"long_ratio": 0.8, "close": 100.0, "turnover": 1e12})])
|
||
store.write_features("BBBUSDT", [(d * _DAY, {"long_ratio": 0.2, "close": bbb_close[d], "turnover": 2e6})])
|
||
raw = positioning_returns_from_store(store, cost_bps=0.0) # no haircut
|
||
small = positioning_returns_from_store(store, cost_bps=0.0, capacity_capital=1_000.0) # tiny book
|
||
large = positioning_returns_from_store(store, cost_bps=0.0, capacity_capital=100_000_000.0) # huge book
|
||
store.close()
|
||
spike = max(raw, key=lambda d: raw[d])
|
||
assert small[spike] == raw[spike] # tiny capital: position << turnover → fully capturable (cf=1)
|
||
assert large[spike] < 0.25 * raw[spike] # huge capital: the un-exitable gain is heavily capped
|
||
|
||
|
||
def _seed_tail(store: TimescaleFeatureStore, *, spike: float, n: int = 28) -> None:
|
||
"""2 coins with stable contrarian weights + OPPOSITE small daily moves (so the book has a real ~1% daily
|
||
scale for σ_book), and one coin's next-day return spiking on day n-2. Huge turnover → capacity never binds,
|
||
isolating the tail-cap."""
|
||
ac, bc = 100.0, 100.0
|
||
for d in range(n):
|
||
store.write_features("AAAUSDT", [(d * _DAY, {"long_ratio": 0.8, "close": ac, "turnover": 1e12})])
|
||
store.write_features("BBBUSDT", [(d * _DAY, {"long_ratio": 0.2, "close": bc, "turnover": 1e12})])
|
||
ac *= 1.0 + (0.01 if d % 2 else -0.01)
|
||
bc *= 1.0 + ((spike if d == n - 2 else (-0.01 if d % 2 else 0.01)))
|
||
|
||
|
||
def test_tail_cap_clips_dominating_gain_not_loss() -> None:
|
||
from fxhnt.application.bybit_positioning_eval import positioning_returns_from_store
|
||
# a dominating GAIN day (BBB long, +60%) is clipped toward K·σ_book...
|
||
sg = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_tail(sg, spike=0.60)
|
||
raw = positioning_returns_from_store(sg, cost_bps=0.0)
|
||
capped = positioning_returns_from_store(sg, cost_bps=0.0, tail_cap_k=4.0)
|
||
sg.close()
|
||
gday = max(raw, key=lambda d: raw[d])
|
||
assert capped[gday] < 0.5 * raw[gday] # the fat-right-tail day is materially clipped
|
||
# ...but a dominating LOSS day (BBB −40%) is untouched (gains-only: you are stuck with losses).
|
||
sl = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_tail(sl, spike=-0.40)
|
||
raw_l = positioning_returns_from_store(sl, cost_bps=0.0)
|
||
capped_l = positioning_returns_from_store(sl, cost_bps=0.0, tail_cap_k=4.0)
|
||
sl.close()
|
||
lday = min(raw_l, key=lambda d: raw_l[d])
|
||
assert capped_l[lday] == raw_l[lday] # gains-only: the loss day is NOT capped
|
||
|
||
|
||
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. positioning_weights --------------------------------------------------------------------
|
||
|
||
def test_contrarian_weights_short_high_ratio_long_low_ratio() -> None:
|
||
# AAA over-LONG (high buyRatio → SHORT), CCC over-SHORT (low buyRatio → LONG).
|
||
ratio = {0: {"AAAUSDT": 0.70, "BBBUSDT": 0.50, "CCCUSDT": 0.30}}
|
||
w = positioning_weights(ratio)[0]
|
||
assert w["AAAUSDT"] < 0.0 # short the over-long crowd
|
||
assert w["CCCUSDT"] > 0.0 # long the over-short crowd
|
||
assert abs(w["BBBUSDT"]) < 1e-9 # the at-mean coin carries ~zero weight
|
||
|
||
|
||
def test_contrarian_weights_market_neutral_and_unit_gross() -> None:
|
||
ratio = {0: {"AAAUSDT": 0.80, "BBBUSDT": 0.60, "CCCUSDT": 0.40, "DDDUSDT": 0.20}}
|
||
w = positioning_weights(ratio)[0]
|
||
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_momentum_weights_are_exact_negation_of_contrarian() -> None:
|
||
ratio = {0: {"AAAUSDT": 0.80, "BBBUSDT": 0.60, "CCCUSDT": 0.40, "DDDUSDT": 0.20}}
|
||
w_con = positioning_weights(ratio, direction="contrarian")[0]
|
||
w_mom = positioning_weights(ratio, direction="momentum")[0]
|
||
assert set(w_con) == set(w_mom)
|
||
for c in w_con:
|
||
assert math.isclose(w_mom[c], -w_con[c], abs_tol=1e-12)
|
||
# momentum LONGS the over-long crowd (informed-flow continuation).
|
||
assert w_mom["AAAUSDT"] > 0.0
|
||
assert w_mom["DDDUSDT"] < 0.0
|
||
|
||
|
||
def test_default_direction_is_contrarian() -> None:
|
||
ratio = {0: {"AAAUSDT": 0.7, "BBBUSDT": 0.3}}
|
||
assert positioning_weights(ratio) == positioning_weights(ratio, direction="contrarian")
|
||
|
||
|
||
def test_weights_skip_day_with_fewer_than_two_coins() -> None:
|
||
ratio = {0: {"AAAUSDT": 0.7}, 1: {"AAAUSDT": 0.6, "BBBUSDT": 0.4}}
|
||
out = positioning_weights(ratio)
|
||
assert 0 not in out and 1 in out
|
||
|
||
|
||
def test_coin_gross_cap_clips_and_renormalizes() -> None:
|
||
# Realistic ~60-coin panel with one dominant over-long coin (LABUSDT-class). Uncapped, the dominant coin's
|
||
# |weight| exceeds the 0.07 cap; capped, it is pulled down close to the cap (single-pass clip+renormalize
|
||
# binds strongly when many coins carry weight). Verifies the cap substantially reduces concentration,
|
||
# preserves unit gross, and keeps the sign structure.
|
||
ratio = {0: {f"C{i}USDT": 0.30 + 0.0067 * i for i in range(60)}} # 60 coins spread from 0.30 to ~0.70
|
||
ratio[0]["DOMUSDT"] = 1.10 # the extreme over-long coin (fade -> short)
|
||
uncapped = positioning_weights(ratio)[0]
|
||
capped = positioning_weights(ratio, coin_gross_cap=0.07)[0]
|
||
assert abs(uncapped["DOMUSDT"]) > 0.07 # dominant coin exceeds the cap uncapped
|
||
assert abs(capped["DOMUSDT"]) < abs(uncapped["DOMUSDT"]) # cap pulls it down
|
||
assert abs(capped["DOMUSDT"]) < 0.085 # close to the 0.07 cap (single-pass slack)
|
||
assert abs(sum(abs(w) for w in capped.values()) - 1.0) < 1e-9 # still unit gross
|
||
assert capped["DOMUSDT"] < 0.0 # over-long crowd -> short (sign preserved)
|
||
|
||
|
||
def test_coin_gross_cap_none_is_byte_identical() -> None:
|
||
ratio = {0: {"AAAUSDT": 0.9, "BBBUSDT": 0.5, "CCCUSDT": 0.1}}
|
||
assert positioning_weights(ratio, coin_gross_cap=None) == positioning_weights(ratio)
|
||
|
||
|
||
# --- 2. positioning_metrics_from_store ---------------------------------------------------------
|
||
|
||
def _seed_contrarian(store: TimescaleFeatureStore, *, days: int = 80) -> None:
|
||
"""Seed `store` so FADING retail is profitable: when AAA is over-LONG (high buyRatio) its price FALLS
|
||
next day (the crowd is wrong) — so SHORTing it earns. Two coins, oscillating positioning, flat-ish drift.
|
||
The price return is close-to-close; the contrarian bet is on price reversion."""
|
||
px = {"AAAUSDT": 100.0, "BBBUSDT": 100.0}
|
||
for d in range(days):
|
||
crowd_long_a = (d % 2 == 0) # AAA over-long on even days, BBB on odd days
|
||
ra = 0.70 if crowd_long_a else 0.30
|
||
rb = 0.30 if crowd_long_a else 0.70
|
||
store.write_features("AAAUSDT", [(d * _DAY, {"long_ratio": ra, "close": px["AAAUSDT"]})])
|
||
store.write_features("BBBUSDT", [(d * _DAY, {"long_ratio": rb, "close": px["BBBUSDT"]})])
|
||
# The over-long coin's price FALLS next day (crowd wrong → reversion); the over-short coin rises.
|
||
px["AAAUSDT"] *= (1.0 - 0.004) if crowd_long_a else (1.0 + 0.004)
|
||
px["BBBUSDT"] *= (1.0 + 0.004) if crowd_long_a else (1.0 - 0.004)
|
||
# One extra day so the last weighted day has a realized next-day return.
|
||
d = days
|
||
store.write_features("AAAUSDT", [(d * _DAY, {"long_ratio": 0.5, "close": px["AAAUSDT"]})])
|
||
store.write_features("BBBUSDT", [(d * _DAY, {"long_ratio": 0.5, "close": px["BBBUSDT"]})])
|
||
|
||
|
||
def test_metrics_contrarian_positive_when_fading_retail_pays() -> None:
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_contrarian(store)
|
||
m = positioning_metrics_from_store(store, cost_bps=0.0, direction="contrarian")
|
||
store.close()
|
||
assert m["available"] is True
|
||
assert m["n_coins"] == 2
|
||
assert m["days"] > 0
|
||
assert math.isfinite(m["sharpe"])
|
||
assert m["total_return"] > 0.0 # fading the (wrong) crowd is profitable gross
|
||
|
||
|
||
def test_metrics_direction_flip_makes_momentum_lose() -> None:
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_contrarian(store)
|
||
m_con = positioning_metrics_from_store(store, cost_bps=0.0, direction="contrarian")
|
||
m_mom = positioning_metrics_from_store(store, cost_bps=0.0, direction="momentum")
|
||
store.close()
|
||
assert m_mom["sharpe"] < 0.0 < m_con["sharpe"]
|
||
|
||
|
||
def test_metrics_na_when_no_ratio_data() -> None:
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
store.write_features("AAAUSDT", [(0, {"close": 100.0})]) # close but no long_ratio
|
||
m = positioning_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_contrarian(inner)
|
||
store = _WriteTripwireStore(inner)
|
||
m = positioning_metrics_from_store(store, cost_bps=5.5) # must not trip the tripwire
|
||
inner.close()
|
||
assert m["available"] is True
|
||
|
||
|
||
# --- 3. verify diagnostics ---------------------------------------------------------------------
|
||
|
||
def _seed_momentum(store: TimescaleFeatureStore, *, days: int = 220) -> None:
|
||
"""Seed `store` so the MOMENTUM (informed-flow) direction is profitable across MULTIPLE years and broad
|
||
across 3 coins: the over-LONG coin keeps RISING (flow continues). Start near 2021-01-01 so a 220-day span
|
||
crosses calendar years for the per-year slicing."""
|
||
px = {"AAAUSDT": 100.0, "BBBUSDT": 100.0, "CCCUSDT": 100.0}
|
||
start_day = 18628 # 2021-01-01
|
||
for i in range(days):
|
||
d = start_day + i
|
||
crowd = ["AAAUSDT", "BBBUSDT", "CCCUSDT"][i % 3] # rotate the over-long coin (broad attribution)
|
||
for c in ("AAAUSDT", "BBBUSDT", "CCCUSDT"):
|
||
r = 0.70 if c == crowd else 0.40
|
||
store.write_features(c, [(d * _DAY, {"long_ratio": r, "close": px[c]})])
|
||
# CONTINUATION: the over-long perp keeps rising; the others drift down.
|
||
for c in ("AAAUSDT", "BBBUSDT", "CCCUSDT"):
|
||
px[c] *= (1.0 + 0.005) if c == crowd else (1.0 - 0.0025)
|
||
d = start_day + days
|
||
for c in ("AAAUSDT", "BBBUSDT", "CCCUSDT"):
|
||
store.write_features(c, [(d * _DAY, {"long_ratio": 0.5, "close": px[c]})])
|
||
|
||
|
||
def test_verify_momentum_beats_contrarian_on_continuation_fixture() -> None:
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_momentum(store)
|
||
rep = verify_positioning_edge(store, cost_bps=5.5)
|
||
store.close()
|
||
assert rep["available"] is True
|
||
mom = rep["net_cost"]["momentum"][5.5]
|
||
con = rep["net_cost"]["contrarian"][5.5]
|
||
assert mom["sharpe"] > 0.0 > con["sharpe"]
|
||
|
||
|
||
def test_verify_cost_sensitivity_is_monotone() -> None:
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_momentum(store)
|
||
rep = verify_positioning_edge(store, cost_bps=5.5, cost_grid=(5.5, 11.0, 22.0))
|
||
store.close()
|
||
mom = rep["net_cost"]["momentum"]
|
||
assert mom[5.5]["sharpe"] >= mom[11.0]["sharpe"] >= mom[22.0]["sharpe"]
|
||
|
||
|
||
def test_verify_per_coin_attribution_sums_to_total() -> None:
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_momentum(store)
|
||
rep = verify_positioning_edge(store, cost_bps=0.0)
|
||
store.close()
|
||
attr = rep["per_coin"]["momentum"]
|
||
total = rep["gross_total"]["momentum"]
|
||
assert math.isclose(sum(attr.values()), total, rel_tol=1e-9, abs_tol=1e-12)
|
||
assert len(attr) == 3
|
||
assert 0.0 <= rep["top5_share"]["momentum"] <= 1.0 + 1e-9
|
||
|
||
|
||
def test_verify_per_year_sharpes_on_right_slices() -> None:
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_momentum(store, days=400) # spans 2021 and 2022
|
||
rep = verify_positioning_edge(store, cost_bps=5.5)
|
||
store.close()
|
||
per_year = rep["per_year"]["momentum"]
|
||
assert "2021" in per_year and "2022" in per_year
|
||
total_days = rep["net_cost"]["momentum"][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_momentum(store)
|
||
rep = verify_positioning_edge(store, cost_bps=5.5)
|
||
store.close()
|
||
assert rep["avg_turnover"]["momentum"] > 0.0
|
||
assert rep["avg_turnover"]["contrarian"] > 0.0
|
||
|
||
|
||
def test_verify_reports_correlation_with_all_three_edges() -> None:
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_momentum(store)
|
||
rep = verify_positioning_edge(store, cost_bps=5.5)
|
||
store.close()
|
||
# The corr block carries a key per direction, each mapping each existing edge -> corr-or-None.
|
||
for d in ("contrarian", "momentum"):
|
||
corrs = rep["corr_edges"][d]
|
||
for edge in ("tstrend", "unlock", "xsfunding"):
|
||
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_momentum(inner)
|
||
store = _WriteTripwireStore(inner)
|
||
rep = verify_positioning_edge(store, cost_bps=5.5)
|
||
inner.close()
|
||
assert rep["available"] is True
|
||
|
||
|
||
# --- 4. CLI smoke ------------------------------------------------------------------------------
|
||
|
||
def test_cli_positioning_eval_prints_metrics(monkeypatch) -> None:
|
||
from typer.testing import CliRunner
|
||
|
||
import fxhnt.cli as cli
|
||
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_contrarian(store)
|
||
|
||
monkeypatch.setattr(
|
||
"fxhnt.adapters.warehouse.timescale_feature_store.TimescaleFeatureStore",
|
||
lambda *_a, **_k: store, raising=False)
|
||
|
||
result = CliRunner().invoke(cli.app, ["bybit-positioning-eval", "--all"])
|
||
store.close()
|
||
assert result.exit_code == 0, result.output
|
||
out = result.output.lower()
|
||
assert "positioning" in out
|
||
assert "contrarian" in out
|
||
|
||
|
||
def test_cli_positioning_eval_verify_prints_diagnostics(monkeypatch) -> None:
|
||
from typer.testing import CliRunner
|
||
|
||
import fxhnt.cli as cli
|
||
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
_seed_momentum(store)
|
||
|
||
monkeypatch.setattr(
|
||
"fxhnt.adapters.warehouse.timescale_feature_store.TimescaleFeatureStore",
|
||
lambda *_a, **_k: store, raising=False)
|
||
|
||
result = CliRunner().invoke(cli.app, ["bybit-positioning-eval", "--all", "--verify"])
|
||
store.close()
|
||
assert result.exit_code == 0, result.output
|
||
out = result.output.lower()
|
||
assert "contrarian" 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
|
||
|
||
|
||
# --- 5. coin_gross_cap constant and parameter threading through book_breakdown + returns_from_store ------
|
||
|
||
def test_positioning_coin_gross_cap_constant_is_007() -> None:
|
||
from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP
|
||
assert POSITIONING_COIN_GROSS_CAP == 0.07
|
||
|
||
|
||
def test_coin_gross_cap_reduces_single_coin_loss_in_book() -> None:
|
||
"""Shock store: one coin (retail over-short at extreme long_ratio ~0.05) craters ~50% the next day.
|
||
Uncapped, that coin's loss dominates the book loss. Capped at 0.07, its weight is bounded,
|
||
reducing (making less bad) the worst-day loss."""
|
||
from fxhnt.application.bybit_positioning_eval import positioning_returns_from_store
|
||
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
|
||
# Build shock store: 4 coins, 1 at extreme long_ratio (retail over-short), crash next day
|
||
# Days 0-2: normal, Day 2->3: crash day for LABUSDT
|
||
px = {"LABUSDT": 100.0, "AAAUSDT": 100.0, "BBBUSDT": 100.0, "CCCUSDT": 100.0}
|
||
for d in range(4):
|
||
# LABUSDT is over-short (low long_ratio ~0.05), others are near median ~0.5
|
||
store.write_features("LABUSDT", [(d * _DAY, {"long_ratio": 0.05, "close": px["LABUSDT"]})])
|
||
store.write_features("AAAUSDT", [(d * _DAY, {"long_ratio": 0.50, "close": px["AAAUSDT"]})])
|
||
store.write_features("BBBUSDT", [(d * _DAY, {"long_ratio": 0.48, "close": px["BBBUSDT"]})])
|
||
store.write_features("CCCUSDT", [(d * _DAY, {"long_ratio": 0.52, "close": px["CCCUSDT"]})])
|
||
# Normal prices on days 0, 1, 2
|
||
if d < 2:
|
||
px["LABUSDT"] *= 1.00
|
||
px["AAAUSDT"] *= 1.00
|
||
px["BBBUSDT"] *= 1.00
|
||
px["CCCUSDT"] *= 1.00
|
||
# LABUSDT crashes ~50% on the transition day 2->3 (its long_ratio on day 2 forms the weight)
|
||
elif d == 2:
|
||
px["LABUSDT"] *= 0.50 # crash next day
|
||
px["AAAUSDT"] *= 1.00
|
||
px["BBBUSDT"] *= 1.00
|
||
px["CCCUSDT"] *= 1.00
|
||
# (day 3: no more transitions)
|
||
|
||
uncapped = positioning_returns_from_store(store, cost_bps=0.0)
|
||
capped = positioning_returns_from_store(store, cost_bps=0.0, coin_gross_cap=0.07)
|
||
store.close()
|
||
|
||
# The store MUST produce a book (empty dicts would mean the fixture is broken — assert loudly, never
|
||
# skip the real assertion silently).
|
||
assert uncapped and capped, "shock store produced no positioning returns — fixture broken"
|
||
# Worst day = the crash day (day 2: weight formed, day 2->3 loss realized). The cap bounds the per-coin
|
||
# weight, making the worst-day loss less bad (less negative).
|
||
shock_day = min(uncapped, key=lambda d: uncapped[d])
|
||
assert capped[shock_day] > uncapped[shock_day], \
|
||
f"cap should improve worst day: uncapped={uncapped[shock_day]}, capped={capped[shock_day]}"
|
||
|
||
|
||
def test_sleeve_returns_positioning_forwards_coin_gross_cap() -> None:
|
||
"""sleeve_returns_from_store forwards coin_gross_cap to the positioning edge only."""
|
||
from fxhnt.application.bybit_book_eval import sleeve_returns_from_store
|
||
|
||
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
||
|
||
# Build shock store: 4 coins, 1 at extreme long_ratio (retail over-short), crash next day
|
||
px = {"LABUSDT": 100.0, "AAAUSDT": 100.0, "BBBUSDT": 100.0, "CCCUSDT": 100.0}
|
||
for d in range(4):
|
||
store.write_features("LABUSDT", [(d * _DAY, {"long_ratio": 0.05, "close": px["LABUSDT"]})])
|
||
store.write_features("AAAUSDT", [(d * _DAY, {"long_ratio": 0.50, "close": px["AAAUSDT"]})])
|
||
store.write_features("BBBUSDT", [(d * _DAY, {"long_ratio": 0.48, "close": px["BBBUSDT"]})])
|
||
store.write_features("CCCUSDT", [(d * _DAY, {"long_ratio": 0.52, "close": px["CCCUSDT"]})])
|
||
if d < 2:
|
||
px["LABUSDT"] *= 1.00
|
||
px["AAAUSDT"] *= 1.00
|
||
px["BBBUSDT"] *= 1.00
|
||
px["CCCUSDT"] *= 1.00
|
||
elif d == 2:
|
||
px["LABUSDT"] *= 0.50
|
||
px["AAAUSDT"] *= 1.00
|
||
px["BBBUSDT"] *= 1.00
|
||
px["CCCUSDT"] *= 1.00
|
||
|
||
uncapped = sleeve_returns_from_store(store, "positioning", cost_bps=0.0)
|
||
capped = sleeve_returns_from_store(store, "positioning", cost_bps=0.0, coin_gross_cap=0.07)
|
||
store.close()
|
||
|
||
# The store MUST produce a book (empty dicts would mean the seam or fixture is broken — assert loudly,
|
||
# never skip the real assertion silently).
|
||
assert uncapped and capped, "shock store produced no positioning returns via the sleeve seam — broken"
|
||
# cap forwarded through the sleeve seam -> worst day STRICTLY improved (the deterministic shock always
|
||
# yields a strict improvement: −0.25 -> −0.125). Strict `>` so a silent no-op (cap not forwarded) fails.
|
||
shock_day = min(uncapped, key=lambda d: uncapped[d])
|
||
assert capped[shock_day] > uncapped[shock_day], \
|
||
f"cap should improve worst day: uncapped={uncapped[shock_day]}, capped={capped[shock_day]}"
|