From 1ec5afa0c47c4a408554959d5f322e17d276954f Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 17:44:42 +0000 Subject: [PATCH 01/15] feat(positioning): ex-ante per-coin gross cap in positioning_weights (default off) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds coin_gross_cap parameter to positioning_weights() to limit concentration in any single coin. When set, clips each weight to ±cap then renormalizes to unit gross (Σ|w|=1). When None (default), behavior is byte-identical to before. Tests verify: - cap clips weight concentration and maintains unit gross - None cap produces byte-identical output - All existing positioning tests still pass Co-Authored-By: Claude Opus 4.8 --- .../application/bybit_positioning_eval.py | 19 ++++++++++++++--- .../test_bybit_positioning_eval.py | 21 +++++++++++++++++++ 2 files changed, 37 insertions(+), 3 deletions(-) diff --git a/src/fxhnt/application/bybit_positioning_eval.py b/src/fxhnt/application/bybit_positioning_eval.py index 35d306f..d5ea021 100644 --- a/src/fxhnt/application/bybit_positioning_eval.py +++ b/src/fxhnt/application/bybit_positioning_eval.py @@ -76,7 +76,8 @@ def _ratio_by_day(long_ratio_panel: dict[str, dict[int, float]]) -> dict[int, di # --- 2. the cross-sectional, market-neutral weights -------------------------------------------- def positioning_weights(ratio_by_day: dict[int, dict[str, float]], *, - direction: str = "contrarian") -> dict[int, dict[str, float]]: + direction: str = "contrarian", + coin_gross_cap: float | None = None) -> dict[int, dict[str, float]]: """`{day: {coin: w}}` market-neutral cross-sectional weights from the long/short positioning signal. `direction="contrarian"` (DEFAULT) SHORTs the high-buyRatio coins (retail over-long → fade) and LONGs the @@ -84,7 +85,12 @@ def positioning_weights(ratio_by_day: dict[int, dict[str, float]], *, / SHORT the over-short (informed-flow continuation). Construction per day: 1. DEMEAN the per-day buyRatio across coins (so the book is dollar/market-neutral, Σw ≈ 0); 2. apply the DIRECTION sign — contrarian = −demeaned (short the rich/over-long), momentum = +demeaned; - 3. normalise to UNIT GROSS (Σ|w| = 1). + 3. normalise to UNIT GROSS (Σ|w| = 1); + 4. (OPTIONAL) if `coin_gross_cap` is set, CLIP each |w| to the cap and RENORMALISE to unit gross — + an EX-ANTE per-coin concentration limit (applied to the FORMING weight, so it is strictly causal: + it cannot see the coin's realized return). Validated at 0.07 (full-history sweep: Sharpe 2.60→2.69, + maxDD −32.3%→−29.9%, per-year non-decreasing) — caps the LABUSDT-class implosions (a 10.3% long leg + that did −59%/−79.5% on 2026-07-06/07) without over-throttling the edge (0.04 falsified: Sharpe→2.43). Because the gross is identical for both directions, `momentum` weights are the exact per-coin negation of `contrarian`. A day with <2 coins, or zero gross after demeaning, yields no weights (skipped). Pure.""" @@ -100,7 +106,14 @@ def positioning_weights(ratio_by_day: dict[int, dict[str, float]], *, gross = sum(abs(w) for w in raw.values()) if gross <= 0.0: continue - out[int(day)] = {c: w / gross for c, w in raw.items()} + w = {c: x / gross for c, x in raw.items()} + if coin_gross_cap is not None: + w = {c: max(-coin_gross_cap, min(coin_gross_cap, x)) for c, x in w.items()} + capped_gross = sum(abs(x) for x in w.values()) + if capped_gross <= 0.0: + continue + w = {c: x / capped_gross for c, x in w.items()} + out[int(day)] = w return out diff --git a/tests/integration/test_bybit_positioning_eval.py b/tests/integration/test_bybit_positioning_eval.py index 8bbfb30..1f0b635 100644 --- a/tests/integration/test_bybit_positioning_eval.py +++ b/tests/integration/test_bybit_positioning_eval.py @@ -152,6 +152,27 @@ def test_weights_skip_day_with_fewer_than_two_coins() -> None: assert 0 not in out and 1 in out +def test_coin_gross_cap_clips_and_renormalizes() -> None: + # One coin dominates (retail extreme) -> uncapped weight is large; cap must clip + renormalize + # to reduce concentration. The cap reduces the max weight from dominating level to near-cap level. + ratio = {0: {f"C{i}USDT": 0.50 for i in range(100)}} # 100 neutral coins + ratio[0]["DOMUSDT"] = 0.95 # Add the extreme long coin + uncapped = positioning_weights(ratio)[0] + max_uncapped = max(abs(w) for w in uncapped.values()) + assert max_uncapped > 0.40 # DOM dominates uncapped + capped = positioning_weights(ratio, coin_gross_cap=0.40)[0] + max_capped = max(abs(w) for w in capped.values()) + assert max_capped < max_uncapped # cap reduces the maximum weight + assert abs(sum(abs(w) for w in capped.values()) - 1.0) < 1e-9 # still unit gross + # sign structure preserved: DOM still short (over-long crowd) + assert capped["DOMUSDT"] < 0.0 + + +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: From f8831a7502c32adf35b6c8346f30f650026a8c26 Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 17:52:22 +0000 Subject: [PATCH 02/15] test(positioning): realistic ~60-coin panel proves coin_gross_cap binds at 0.07 --- .../test_bybit_positioning_eval.py | 24 +++++++++---------- 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/tests/integration/test_bybit_positioning_eval.py b/tests/integration/test_bybit_positioning_eval.py index 1f0b635..4a16c74 100644 --- a/tests/integration/test_bybit_positioning_eval.py +++ b/tests/integration/test_bybit_positioning_eval.py @@ -153,19 +153,19 @@ def test_weights_skip_day_with_fewer_than_two_coins() -> None: def test_coin_gross_cap_clips_and_renormalizes() -> None: - # One coin dominates (retail extreme) -> uncapped weight is large; cap must clip + renormalize - # to reduce concentration. The cap reduces the max weight from dominating level to near-cap level. - ratio = {0: {f"C{i}USDT": 0.50 for i in range(100)}} # 100 neutral coins - ratio[0]["DOMUSDT"] = 0.95 # Add the extreme long coin + # 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] - max_uncapped = max(abs(w) for w in uncapped.values()) - assert max_uncapped > 0.40 # DOM dominates uncapped - capped = positioning_weights(ratio, coin_gross_cap=0.40)[0] - max_capped = max(abs(w) for w in capped.values()) - assert max_capped < max_uncapped # cap reduces the maximum weight - assert abs(sum(abs(w) for w in capped.values()) - 1.0) < 1e-9 # still unit gross - # sign structure preserved: DOM still short (over-long crowd) - assert capped["DOMUSDT"] < 0.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: From a455dc0fead96e34956905aa35cca8ff146031e6 Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 18:00:20 +0000 Subject: [PATCH 03/15] feat(positioning): thread coin_gross_cap through book_breakdown + returns_from_store - Add coin_gross_cap: float | None = None parameter to _book_breakdown signature - Pass coin_gross_cap to positioning_weights call in _book_breakdown - Add coin_gross_cap: float | None = None parameter to positioning_returns_from_store signature - Pass coin_gross_cap through to _book_breakdown in positioning_returns_from_store - Add docstring note explaining the ex-ante per-coin concentration cap - Add test_coin_gross_cap_reduces_single_coin_loss_in_book: validates that the cap reduces worst-day loss on a shock store with one dominating coin crash All existing tests remain green (24 passed); with coin_gross_cap=None (default), behavior is byte-identical to prior behavior. Co-Authored-By: Claude Opus 4.8 --- .../application/bybit_positioning_eval.py | 13 ++++-- .../test_bybit_positioning_eval.py | 45 +++++++++++++++++++ 2 files changed, 54 insertions(+), 4 deletions(-) diff --git a/src/fxhnt/application/bybit_positioning_eval.py b/src/fxhnt/application/bybit_positioning_eval.py index d5ea021..759b0cf 100644 --- a/src/fxhnt/application/bybit_positioning_eval.py +++ b/src/fxhnt/application/bybit_positioning_eval.py @@ -153,7 +153,8 @@ def _robust_sigma(vals: list[float]) -> float: def _book_breakdown(store: _FeatureStore, *, universe: set[str] | None, cost_bps: float, direction: str, capacity_capital: float | None = None, participation: float = 0.10, turnover_window: int = 30, - tail_cap_k: float | None = None, book_vol_window: int = 60) -> dict[str, Any]: + tail_cap_k: float | None = None, book_vol_window: int = 60, + coin_gross_cap: float | None = None) -> dict[str, Any]: """READ-ONLY, memory-bounded per-day book breakdown for the positioning edge in ONE pass. Returns: @@ -187,7 +188,8 @@ def _book_breakdown(store: _FeatureStore, *, universe: set[str] | None, cost_bps if not close_panel: return empty - weights_by_day = positioning_weights(_ratio_by_day(long_ratio), direction=direction) + weights_by_day = positioning_weights(_ratio_by_day(long_ratio), direction=direction, + coin_gross_cap=coin_gross_cap) if not weights_by_day: return empty next_ret = _next_day_returns(close_panel) @@ -301,6 +303,7 @@ def positioning_returns_from_store(store: _FeatureStore, *, universe: set[str] | cost_bps: float = 5.5, direction: str = "contrarian", capacity_capital: float | None = None, tail_cap_k: float | None = None, + coin_gross_cap: float | None = None, ) -> dict[int, float]: """READ-ONLY, memory-bounded per-day RETURN SERIES `{epoch_day: ret}` for the positioning edge — the book-level seam the 4th sleeve consumes (mirrors how `bybit_book_eval.sleeve_returns_from_store` exposes @@ -308,9 +311,11 @@ def positioning_returns_from_store(store: _FeatureStore, *, universe: set[str] | `"momentum"` is the sign-flip. Reads restricted to `universe` (the memory bound). Writes NOTHING. `capacity_capital` / `tail_cap_k` enable the gains-only capacity + tail-concentration haircuts (see - `_book_breakdown`); both None (default) → the raw capturable-unaware book, unchanged.""" + `_book_breakdown`); both None (default) → the raw capturable-unaware book, unchanged. + `coin_gross_cap` (None default) applies the ex-ante per-coin concentration cap (see `positioning_weights`).""" return _book_breakdown(store, universe=universe, cost_bps=cost_bps, direction=direction, - capacity_capital=capacity_capital, tail_cap_k=tail_cap_k)["net"] + capacity_capital=capacity_capital, tail_cap_k=tail_cap_k, + coin_gross_cap=coin_gross_cap)["net"] # Back-compat alias: `positioning_returns` was the original name; keep it pointing at the canonical diff --git a/tests/integration/test_bybit_positioning_eval.py b/tests/integration/test_bybit_positioning_eval.py index 4a16c74..5ef8489 100644 --- a/tests/integration/test_bybit_positioning_eval.py +++ b/tests/integration/test_bybit_positioning_eval.py @@ -373,3 +373,48 @@ def test_cli_positioning_eval_verify_prints_diagnostics(monkeypatch) -> None: 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 parameter threading through book_breakdown + returns_from_store ------ + +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() + + # Worst day should be the crash day (day 2: weight formed, day 2->3 loss realized) + if uncapped and capped: + shock_day = min(uncapped, key=lambda d: uncapped[d]) + # The cap bounds the per-coin weight, making the worst-day loss less bad (less negative) + assert capped[shock_day] > uncapped[shock_day], \ + f"cap should improve worst day: uncapped={uncapped[shock_day]}, capped={capped[shock_day]}" From 9e94b3b64591dbd1f1a3fa887607299d8d9a748f Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 18:14:08 +0000 Subject: [PATCH 04/15] feat(positioning): forward coin_gross_cap through sleeve_returns_from_store Thread coin_gross_cap parameter through _positioning_series and sleeve_returns_from_store to enable per-coin weight capping at the book level. The cap is forwarded only to the positioning edge; other sleeves (tstrend, unlock, xsfunding, stablecoin) remain unchanged. With coin_gross_cap=None (default), behavior is byte-identical to the previous version. Co-Authored-By: Claude Opus 4.8 --- src/fxhnt/application/bybit_book_eval.py | 15 +++++--- .../test_bybit_positioning_eval.py | 35 +++++++++++++++++++ 2 files changed, 45 insertions(+), 5 deletions(-) diff --git a/src/fxhnt/application/bybit_book_eval.py b/src/fxhnt/application/bybit_book_eval.py index 9784768..f82552d 100644 --- a/src/fxhnt/application/bybit_book_eval.py +++ b/src/fxhnt/application/bybit_book_eval.py @@ -225,7 +225,7 @@ def _xsfunding_series(store: _FeatureStore, *, cost_bps: float) -> dict[int, flo def _positioning_series(store: _FeatureStore, *, universe: set[str] | None, cost_bps: float, direction: str = "contrarian", capacity_capital: float | None = None, - tail_cap_k: float | None = None) -> dict[int, float]: + tail_cap_k: float | None = None, coin_gross_cap: float | None = None) -> dict[int, float]: """positioning (contrarian long/short account-ratio) per-day return series via the live `bybit_positioning_eval.positioning_returns_from_store`. It reads the `long_ratio` + `close` features through the SAME `_UniverseRestrictedStore` seam (a `WHERE symbol IN (...)` read), so it is memory-bounded @@ -233,11 +233,13 @@ def _positioning_series(store: _FeatureStore, *, universe: set[str] | None, cost is local to avoid a circular import (`bybit_positioning_eval` imports from this module). `capacity_capital` / `tail_cap_k` (both None = off) enable the gains-only capacity + tail-concentration - haircuts that make positioning's booked return capturable + robust (see `_book_breakdown`).""" + haircuts that make positioning's booked return capturable + robust (see `_book_breakdown`). + `coin_gross_cap` (None = off) caps the per-coin weight to improve concentration robustness.""" from fxhnt.application.bybit_positioning_eval import positioning_returns_from_store return positioning_returns_from_store(store, universe=universe, cost_bps=cost_bps, direction=direction, - capacity_capital=capacity_capital, tail_cap_k=tail_cap_k) + capacity_capital=capacity_capital, tail_cap_k=tail_cap_k, + coin_gross_cap=coin_gross_cap) def sleeve_returns_from_store(store: _FeatureStore, edge: str, *, universe: set[str] | None = None, @@ -246,6 +248,7 @@ def sleeve_returns_from_store(store: _FeatureStore, edge: str, *, universe: set[ stablecoin_spot_panel: dict[str, dict[int, float]] | None = None, capacity_capital: float | None = None, tail_cap_k: float | None = None, + coin_gross_cap: float | None = None, ) -> dict[int, float]: """READ-ONLY, memory-bounded per-day RETURN SERIES `{epoch_day: ret}` for ONE edge on `store`. @@ -272,9 +275,11 @@ def sleeve_returns_from_store(store: _FeatureStore, edge: str, *, universe: set[ # positioning_returns_from_store does its OWN _bounded_store(universe) wrap (it also reads the # `long_ratio` panel via read_panel), so pass the RAW store + universe — not the already-bounded # proxy — to avoid double-wrapping. capacity_capital/tail_cap_k thread the positioning haircuts - # (ignored by the other sleeves, which have their own constructions). + # (ignored by the other sleeves, which have their own constructions). coin_gross_cap threads the + # per-coin weight cap (also positioning-only). return _positioning_series(store, universe=universe, cost_bps=cost_bps, - capacity_capital=capacity_capital, tail_cap_k=tail_cap_k) + capacity_capital=capacity_capital, tail_cap_k=tail_cap_k, + coin_gross_cap=coin_gross_cap) raise ValueError( f"unknown edge {edge!r} (known: tstrend, unlock, stablecoin_rotation, xsfunding, positioning)") diff --git a/tests/integration/test_bybit_positioning_eval.py b/tests/integration/test_bybit_positioning_eval.py index 5ef8489..af0b400 100644 --- a/tests/integration/test_bybit_positioning_eval.py +++ b/tests/integration/test_bybit_positioning_eval.py @@ -418,3 +418,38 @@ def test_coin_gross_cap_reduces_single_coin_loss_in_book() -> None: # The cap bounds the per-coin weight, making the worst-day loss less bad (less negative) 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() + + # cap forwarded -> worst day improved (less negative or equal) + if uncapped and capped: + 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]}" From 6a109bfec7003654e5742260083694d78d266505 Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 18:22:12 +0000 Subject: [PATCH 05/15] =?UTF-8?q?test(positioning):=20harden=203=20minors?= =?UTF-8?q?=20=E2=80=94=20assert=20non-empty=20store=20loudly=20+=20strict?= =?UTF-8?q?=20>=20(no=20silent=20no-op)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every minor is a latent bug: the two 'if uncapped and capped:' guards (Task 2+3 tests) would silently pass on an empty store, and the Task 3 '>=' would pass on a no-op forwarding break. Replaced with a loud non-empty assert + strict '>'. Co-Authored-By: Claude Opus 4.8 --- .../test_bybit_positioning_eval.py | 27 +++++++++++-------- 1 file changed, 16 insertions(+), 11 deletions(-) diff --git a/tests/integration/test_bybit_positioning_eval.py b/tests/integration/test_bybit_positioning_eval.py index af0b400..d73ab09 100644 --- a/tests/integration/test_bybit_positioning_eval.py +++ b/tests/integration/test_bybit_positioning_eval.py @@ -412,12 +412,14 @@ def test_coin_gross_cap_reduces_single_coin_loss_in_book() -> None: capped = positioning_returns_from_store(store, cost_bps=0.0, coin_gross_cap=0.07) store.close() - # Worst day should be the crash day (day 2: weight formed, day 2->3 loss realized) - if uncapped and capped: - shock_day = min(uncapped, key=lambda d: uncapped[d]) - # The cap bounds the per-coin weight, making the worst-day loss less bad (less negative) - assert capped[shock_day] > uncapped[shock_day], \ - f"cap should improve worst day: uncapped={uncapped[shock_day]}, capped={capped[shock_day]}" + # 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: @@ -448,8 +450,11 @@ def test_sleeve_returns_positioning_forwards_coin_gross_cap() -> None: capped = sleeve_returns_from_store(store, "positioning", cost_bps=0.0, coin_gross_cap=0.07) store.close() - # cap forwarded -> worst day improved (less negative or equal) - if uncapped and capped: - 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]}" + # 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]}" From 8bbe7fb12366c2ade938424036cb34c5b8a5ed56 Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 18:28:10 +0000 Subject: [PATCH 06/15] feat(positioning): enable coin_gross_cap=0.07 in the live bybit book precompute 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 --- src/fxhnt/application/bybit_book_persist.py | 5 +++-- src/fxhnt/application/bybit_forward_track.py | 5 +++++ tests/integration/test_bybit_positioning_eval.py | 7 ++++++- 3 files changed, 14 insertions(+), 3 deletions(-) diff --git a/src/fxhnt/application/bybit_book_persist.py b/src/fxhnt/application/bybit_book_persist.py index c38ca39..9cea744 100644 --- a/src/fxhnt/application/bybit_book_persist.py +++ b/src/fxhnt/application/bybit_book_persist.py @@ -75,12 +75,13 @@ def _persist_bybit_book(settings, *, min_dollar_vol: float = 10_000_000.0, # haircuts (capacity + tail-cap) are baked into the positioning series HERE so every downstream # reference (bybit_4edge, levered, standalone positioning) reconciles against the capturable book. from fxhnt.application.bybit_book_eval import _DEFAULT_BYBIT_SLEEVES, sleeve_returns_from_store - from fxhnt.application.bybit_forward_track import POSITIONING_TAIL_CAP_K + from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP, POSITIONING_TAIL_CAP_K sleeve_rets = { s: sleeve_returns_from_store(store, s, universe=universe, cost_bps=cost_bps, unlock_events=unlock_events, capacity_capital=settings.paper_capital, - tail_cap_k=POSITIONING_TAIL_CAP_K) + tail_cap_k=POSITIONING_TAIL_CAP_K, + coin_gross_cap=POSITIONING_COIN_GROSS_CAP) for s in _DEFAULT_BYBIT_SLEEVES} repo = PaperRepo(settings.operational_dsn) diff --git a/src/fxhnt/application/bybit_forward_track.py b/src/fxhnt/application/bybit_forward_track.py index 2839cc0..bfe0ce3 100644 --- a/src/fxhnt/application/bybit_forward_track.py +++ b/src/fxhnt/application/bybit_forward_track.py @@ -45,6 +45,11 @@ MAX_BOOK_GROSS = 1.5 # — the adaptivity is in σ_book, not K. The capacity capital is threaded from settings.paper_capital at the # wiring layer (gains-only; barely binds at fund scale, caps thin names as capital grows). POSITIONING_TAIL_CAP_K = 4.0 +# Ex-ante per-coin gross-weight cap for the positioning sleeve — no single coin's |weight| may exceed this +# fraction of the book's unit gross. Validated 2026-07-20 over the full 1295-day history (Sharpe 2.60→2.69, +# maxDD −32.3%→−29.9%, per-year Sharpe non-decreasing every year); caps LABUSDT-class single-coin implosions +# (a 10.3% long leg that did −59%/−79.5% on 2026-07-06/07) without over-throttling (0.04 falsified → 2.43). +POSITIONING_COIN_GROSS_CAP = 0.07 class _FeatureStore(Protocol): diff --git a/tests/integration/test_bybit_positioning_eval.py b/tests/integration/test_bybit_positioning_eval.py index d73ab09..45886b1 100644 --- a/tests/integration/test_bybit_positioning_eval.py +++ b/tests/integration/test_bybit_positioning_eval.py @@ -375,7 +375,12 @@ def test_cli_positioning_eval_verify_prints_diagnostics(monkeypatch) -> None: assert "tstrend" in out and "unlock" in out and "xsfunding" in out -# --- 5. coin_gross_cap parameter threading through book_breakdown + returns_from_store ------ +# --- 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. From 3d1c325d2cff2ed0868f0b122b0f6e44b51e0648 Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 18:40:47 +0000 Subject: [PATCH 07/15] feat(positioning): re-inception positioning+bybit_4edge v3 (coin_gross_cap, backdated 2026-07-06) Bumps definition.version 2->3 for positioning, bybit_4edge, and bybit_4edge_levered (construction changed: Task 4's per-coin gross cap 0.07 on the positioning sleeve). Pins inception_t0 to {"date": "2026-07-06", "version": 3} on all three so the forward anchor recomputes over the existing OOS window instead of resetting the gate to today. bybit_4edge/bybit_4edge_levered move their pin from 2026-07-07 to 2026-07-06 (positioning's date) so the whole book re-warms together from the shock's first day. Updates the existing regression test test_re_homed_sleeves_pin_their_recovered_inception's expected dates for bybit_4edge/bybit_4edge_levered accordingly. --- src/fxhnt/registry.py | 12 ++++++------ tests/unit/test_forward_definition.py | 4 +++- tests/unit/test_positioning_cap_reinception.py | 11 +++++++++++ 3 files changed, 20 insertions(+), 7 deletions(-) create mode 100644 tests/unit/test_positioning_cap_reinception.py diff --git a/src/fxhnt/registry.py b/src/fxhnt/registry.py index 156b1a6..4b9bf8f 100644 --- a/src/fxhnt/registry.py +++ b/src/fxhnt/registry.py @@ -205,8 +205,8 @@ STRATEGY_REGISTRY: dict[str, dict[str, Any]] = { # cost_bps mirrors Settings.bybit_cost_bps; a cost change requires a definition_version bump # to re-inception (else the hashed audit silently diverges from the cost actually charged). "definition": {"params": {"sleeves": ["tstrend", "unlock", "xsfunding", "positioning"], "weighting": "naive-eq", - "cost_bps": 5.5}, "version": 2, "record_mode": "recompute", - "inception_t0": {"date": "2026-07-07", "version": 2}, # pre-re-home OOS start (see xsfunding) + "cost_bps": 5.5}, "version": 3, "record_mode": "recompute", + "inception_t0": {"date": "2026-07-06", "version": 3}, # v3: positioning coin_gross_cap=0.07 (2026-07-20 backfill) "backtest": {"kind": "report"}}, }, # The EXECUTED 4-edge book (sub-project C2): the real Bybit-TESTNET track — its daily return is the @@ -235,8 +235,8 @@ STRATEGY_REGISTRY: dict[str, dict[str, Any]] = { # to re-inception (else the hashed audit silently diverges from the cost actually charged). "definition": {"params": {"sleeves": ["tstrend", "unlock", "xsfunding", "positioning"], "weighting": "carry-static", "gross_max": 1.5, "cost_bps": 5.5}, - "version": 2, "record_mode": "recompute", - "inception_t0": {"date": "2026-07-07", "version": 2}, # pre-re-home OOS start (see xsfunding) + "version": 3, "record_mode": "recompute", + "inception_t0": {"date": "2026-07-06", "version": 3}, # v3: positioning coin_gross_cap=0.07 (2026-07-20 backfill) "backtest": {"kind": "report"}}, }, # The positioning sleeve (contrarian long/short-account-ratio) as its OWN forward-tracked edge — the @@ -253,8 +253,8 @@ STRATEGY_REGISTRY: dict[str, dict[str, Any]] = { "gate_spec": {"gate_type": "reconciliation", "min_forward_days": 14}, # cost_bps mirrors Settings.bybit_cost_bps; a cost change requires a definition_version bump # to re-inception (else the hashed audit silently diverges from the cost actually charged). - "definition": {"params": {"sleeve": "positioning", "cost_bps": 5.5}, "version": 2, "record_mode": "recompute", - "inception_t0": {"date": "2026-07-06", "version": 2}, # pre-re-home OOS start (see xsfunding) + "definition": {"params": {"sleeve": "positioning", "cost_bps": 5.5}, "version": 3, "record_mode": "recompute", + "inception_t0": {"date": "2026-07-06", "version": 3}, # v3: positioning coin_gross_cap=0.07 (2026-07-20 backfill) "backtest": {"kind": "report"}}, }, # NOTE: `xvenue_carry` (cross-venue Bybit↔Deribit static carry) was RETIRED from forward-tracking diff --git a/tests/unit/test_forward_definition.py b/tests/unit/test_forward_definition.py index 997c469..b31d825 100644 --- a/tests/unit/test_forward_definition.py +++ b/tests/unit/test_forward_definition.py @@ -66,7 +66,9 @@ def test_re_homed_sleeves_pin_their_recovered_inception(): # Regression: the Phase-0b faithful re-homes keep their pinned pre-re-home OOS start (recovered 2026-07-15) # so a DB rebuild reproduces the window from git instead of restarting at today. If a sleeve is genuinely # re-tuned later, its version bumps and the (now version-stale) pin is correctly ignored — update it here. + # 2026-07-20: positioning coin_gross_cap=0.07 re-inception (v3) backdated bybit_4edge/bybit_4edge_levered's + # pin from 2026-07-07 to 2026-07-06 (positioning's date, unchanged) so the whole book re-warms together. expected = {"xsfunding": "2026-07-07", "unlock": "2026-06-20", "positioning": "2026-07-06", - "bybit_4edge": "2026-07-07", "bybit_4edge_levered": "2026-07-07"} + "bybit_4edge": "2026-07-06", "bybit_4edge_levered": "2026-07-06"} for sid, date in expected.items(): assert inception_override(sid) == date diff --git a/tests/unit/test_positioning_cap_reinception.py b/tests/unit/test_positioning_cap_reinception.py new file mode 100644 index 0000000..8e67a20 --- /dev/null +++ b/tests/unit/test_positioning_cap_reinception.py @@ -0,0 +1,11 @@ +from fxhnt.registry import STRATEGY_REGISTRY + + +def test_positioning_and_book_bumped_to_v3_backdated(): + for sid in ("positioning", "bybit_4edge", "bybit_4edge_levered"): + d = STRATEGY_REGISTRY[sid]["definition"] + assert d["version"] == 3, f"{sid} version not bumped to 3" + ov = d.get("inception_t0") + assert ov and ov["version"] == 3, f"{sid} inception_t0 not re-pinned to v3" + # backdated to the existing OOS start, NOT reset to today -> the forward window is recomputed, not lost + assert ov["date"] == "2026-07-06" From 46401afbd1adfb5cdc7d279cb8d4f100fae2356e Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 18:51:20 +0000 Subject: [PATCH 08/15] style(positioning): shorten inception_t0 comments to satisfy E501 (my 3 lines) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 3 new inception_t0 comment lines were 137 chars; trimmed to 112. Every minor is a latent bug — a lint violation in the diff is not left for later. Co-Authored-By: Claude Opus 4.8 --- src/fxhnt/registry.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/fxhnt/registry.py b/src/fxhnt/registry.py index 4b9bf8f..a667668 100644 --- a/src/fxhnt/registry.py +++ b/src/fxhnt/registry.py @@ -206,7 +206,7 @@ STRATEGY_REGISTRY: dict[str, dict[str, Any]] = { # to re-inception (else the hashed audit silently diverges from the cost actually charged). "definition": {"params": {"sleeves": ["tstrend", "unlock", "xsfunding", "positioning"], "weighting": "naive-eq", "cost_bps": 5.5}, "version": 3, "record_mode": "recompute", - "inception_t0": {"date": "2026-07-06", "version": 3}, # v3: positioning coin_gross_cap=0.07 (2026-07-20 backfill) + "inception_t0": {"date": "2026-07-06", "version": 3}, # v3: coin_gross_cap=0.07 backfill "backtest": {"kind": "report"}}, }, # The EXECUTED 4-edge book (sub-project C2): the real Bybit-TESTNET track — its daily return is the @@ -236,7 +236,7 @@ STRATEGY_REGISTRY: dict[str, dict[str, Any]] = { "definition": {"params": {"sleeves": ["tstrend", "unlock", "xsfunding", "positioning"], "weighting": "carry-static", "gross_max": 1.5, "cost_bps": 5.5}, "version": 3, "record_mode": "recompute", - "inception_t0": {"date": "2026-07-06", "version": 3}, # v3: positioning coin_gross_cap=0.07 (2026-07-20 backfill) + "inception_t0": {"date": "2026-07-06", "version": 3}, # v3: coin_gross_cap=0.07 backfill "backtest": {"kind": "report"}}, }, # The positioning sleeve (contrarian long/short-account-ratio) as its OWN forward-tracked edge — the @@ -254,7 +254,7 @@ STRATEGY_REGISTRY: dict[str, dict[str, Any]] = { # cost_bps mirrors Settings.bybit_cost_bps; a cost change requires a definition_version bump # to re-inception (else the hashed audit silently diverges from the cost actually charged). "definition": {"params": {"sleeve": "positioning", "cost_bps": 5.5}, "version": 3, "record_mode": "recompute", - "inception_t0": {"date": "2026-07-06", "version": 3}, # v3: positioning coin_gross_cap=0.07 (2026-07-20 backfill) + "inception_t0": {"date": "2026-07-06", "version": 3}, # v3: coin_gross_cap=0.07 backfill "backtest": {"kind": "report"}}, }, # NOTE: `xvenue_carry` (cross-venue Bybit↔Deribit static carry) was RETIRED from forward-tracking From f2730d195d16567926cc2e21367765317bd3fecf Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 18:56:20 +0000 Subject: [PATCH 09/15] fix(types): clear 4 pre-existing mypy errors in the positioning path Every minor is a latent bug, pre-existing is no exemption: - bybit_book_eval __getattr__ missing -> Any return annotation - two unused type: ignore[assignment] comments (mypy: unused) removed - positioning_returns_from_store: bind the dict[int,float] 'net' to a typed local instead of returning Any from _book_breakdown(...)['net'] mypy clean on both files; 26 positioning tests still green (behavior-neutral). Co-Authored-By: Claude Opus 4.8 --- src/fxhnt/application/bybit_book_eval.py | 6 +++--- src/fxhnt/application/bybit_positioning_eval.py | 7 ++++--- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/src/fxhnt/application/bybit_book_eval.py b/src/fxhnt/application/bybit_book_eval.py index f82552d..4112767 100644 --- a/src/fxhnt/application/bybit_book_eval.py +++ b/src/fxhnt/application/bybit_book_eval.py @@ -128,7 +128,7 @@ class _UniverseRestrictedStore: def crypto_spot_panel(self, *, suffix: str = "USDT") -> dict[str, dict[int, float]]: return self._restricted_panel("spot_close", suffix=suffix) - def __getattr__(self, name: str): + def __getattr__(self, name: str) -> Any: # read_panel, close(), and anything else fall through to the inner store unchanged. return getattr(self._inner, name) @@ -212,14 +212,14 @@ def _xsfunding_series(store: _FeatureStore, *, cost_bps: float) -> dict[int, flo captured[int(d)] = rv return orig(daily, days_meta=days_meta) - carry_mod._curve_metrics = _capture # type: ignore[assignment] + carry_mod._curve_metrics = _capture try: # universe=None: the store passed in is ALREADY the universe-restricted proxy, so its panels only # carry the liquid symbols — restricting again here would be a redundant set-intersect over an # already-bounded panel. cost_bps is the per-turnover carry haircut (same as the edge evaluator). carry_metrics_from_store(store, cost_bps=cost_bps, universe=None) finally: - carry_mod._curve_metrics = orig # type: ignore[assignment] + carry_mod._curve_metrics = orig return dict(captured) diff --git a/src/fxhnt/application/bybit_positioning_eval.py b/src/fxhnt/application/bybit_positioning_eval.py index 759b0cf..7c3276b 100644 --- a/src/fxhnt/application/bybit_positioning_eval.py +++ b/src/fxhnt/application/bybit_positioning_eval.py @@ -313,9 +313,10 @@ def positioning_returns_from_store(store: _FeatureStore, *, universe: set[str] | `capacity_capital` / `tail_cap_k` enable the gains-only capacity + tail-concentration haircuts (see `_book_breakdown`); both None (default) → the raw capturable-unaware book, unchanged. `coin_gross_cap` (None default) applies the ex-ante per-coin concentration cap (see `positioning_weights`).""" - return _book_breakdown(store, universe=universe, cost_bps=cost_bps, direction=direction, - capacity_capital=capacity_capital, tail_cap_k=tail_cap_k, - coin_gross_cap=coin_gross_cap)["net"] + net: dict[int, float] = _book_breakdown(store, universe=universe, cost_bps=cost_bps, direction=direction, + capacity_capital=capacity_capital, tail_cap_k=tail_cap_k, + coin_gross_cap=coin_gross_cap)["net"] + return net # Back-compat alias: `positioning_returns` was the original name; keep it pointing at the canonical From 5d4ddcfb39935b575e9d3b457437179dffe5c4a1 Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 19:24:44 +0000 Subject: [PATCH 10/15] fix(positioning): thread coin_gross_cap into the LIVE forward-nav path + attribution (final-review Critical) The per-coin gross cap (0.07) reached the reconciliation REF but not the live forward NAV (BybitFourEdgeStrategy), so the gate would compare a capped ref against an uncapped forward and misread a healthy book as diverging. Threads positioning_coin_gross_cap through BybitFourEdgeStrategy, the four assets.py builders + three nightly call sites, and the three migration_builders.py builders, exactly parallel to the existing positioning_tail_cap_k. Also fixes positioning_weights_and_contributions + sleeve_weights_and_contributions so the paper-book per-symbol attribution reconciles against the capped sleeve return (Sigma weight*return == sleeve_returns_from_store), preserving the SSOT invariant. Default None everywhere (off = byte-identical). --- src/fxhnt/adapters/orchestration/assets.py | 39 ++++++++++++------- .../orchestration/migration_builders.py | 22 ++++++++--- src/fxhnt/application/bybit_forward_track.py | 5 ++- src/fxhnt/application/bybit_paper_weights.py | 9 ++++- .../application/bybit_positioning_eval.py | 4 +- .../test_bybit_paper_reconciliation.py | 37 ++++++++++++++++++ .../test_positioning_forward_track.py | 27 +++++++++++++ 7 files changed, 120 insertions(+), 23 deletions(-) diff --git a/src/fxhnt/adapters/orchestration/assets.py b/src/fxhnt/adapters/orchestration/assets.py index 7179641..a11fc7c 100644 --- a/src/fxhnt/adapters/orchestration/assets.py +++ b/src/fxhnt/adapters/orchestration/assets.py @@ -105,7 +105,8 @@ def refresh_bybit_warehouse(store, *, funding, klines, account_ratio, universe, def build_bybit_4edge_forward_nav(repo, store, *, today, at, universe=None, cost_bps=5.5, unlock_events=None, stablecoin_spot_panel=None, positioning_capacity_capital=None, - positioning_tail_cap_k=None): # type: ignore[no-untyped-def] + positioning_tail_cap_k=None, + positioning_coin_gross_cap=None): # type: ignore[no-untyped-def] """Plain (Dagster-free, unit-testable) snapshot driver for the Bybit 4-edge forward track: build the naive eq-wt 4-sleeve strategy over `store` and run it through the deterministic engine (DB anchor + recompute — `forward_engine.run_track`), returning its result dict. The DB is the SSOT; there is no @@ -119,14 +120,16 @@ def build_bybit_4edge_forward_nav(repo, store, *, today, at, universe=None, cost store, universe=universe, cost_bps=cost_bps, unlock_events=unlock_events, stablecoin_spot_panel=stablecoin_spot_panel, positioning_capacity_capital=positioning_capacity_capital, - positioning_tail_cap_k=positioning_tail_cap_k) + positioning_tail_cap_k=positioning_tail_cap_k, + positioning_coin_gross_cap=positioning_coin_gross_cap) return run_track(repo, "bybit_4edge", strategy, today=today, at=at) def build_bybit_4edge_levered_forward_nav(repo, store, *, today, at, universe=None, cost_bps=5.5, unlock_events=None, stablecoin_spot_panel=None, positioning_capacity_capital=None, - positioning_tail_cap_k=None): # type: ignore[no-untyped-def] + positioning_tail_cap_k=None, + positioning_coin_gross_cap=None): # type: ignore[no-untyped-def] """Observe-only LEVERED shadow of the 4-edge book: the SAME sleeves under the static `_LEVERED_SLEEVE_SHAPE` (carry-weighted, book-gross <= MAX_BOOK_GROSS), run through the deterministic engine (DB anchor + recompute) under its OWN strategy_id `bybit_4edge_levered`. No capital; records what the levered book @@ -140,13 +143,15 @@ def build_bybit_4edge_levered_forward_nav(repo, store, *, today, at, universe=No stablecoin_spot_panel=stablecoin_spot_panel, leverage_shape=_LEVERED_SLEEVE_SHAPE, max_gross=MAX_BOOK_GROSS, positioning_capacity_capital=positioning_capacity_capital, - positioning_tail_cap_k=positioning_tail_cap_k) + positioning_tail_cap_k=positioning_tail_cap_k, + positioning_coin_gross_cap=positioning_coin_gross_cap) return run_track(repo, "bybit_4edge_levered", strategy, today=today, at=at) def build_bybit_sleeve_forward_nav(repo, store, sleeve, *, today, at, universe=None, cost_bps=5.5, unlock_events=None, positioning_capacity_capital=None, - positioning_tail_cap_k=None): # type: ignore[no-untyped-def] + positioning_tail_cap_k=None, + positioning_coin_gross_cap=None): # type: ignore[no-untyped-def] """Plain (Dagster-free, unit-testable) driver for a SINGLE Bybit sleeve as its own forward track: build the one-sleeve `BybitFourEdgeStrategy` over `store` (bybit_features) and run it through the deterministic engine (DB anchor + recompute). The DB (forward_nav + anchor) is the SSOT — no JSON state. Shared by the @@ -155,18 +160,21 @@ def build_bybit_sleeve_forward_nav(repo, store, sleeve, *, today, at, universe=N from fxhnt.application.forward_engine import run_track strategy = BybitFourEdgeStrategy( store, universe=universe, cost_bps=cost_bps, sleeves=[sleeve], unlock_events=unlock_events, - positioning_capacity_capital=positioning_capacity_capital, positioning_tail_cap_k=positioning_tail_cap_k) + positioning_capacity_capital=positioning_capacity_capital, positioning_tail_cap_k=positioning_tail_cap_k, + positioning_coin_gross_cap=positioning_coin_gross_cap) return run_track(repo, sleeve, strategy, today=today, at=at) def build_positioning_forward_nav(repo, store, *, today, at, universe=None, cost_bps=5.5, positioning_capacity_capital=None, - positioning_tail_cap_k=None): # type: ignore[no-untyped-def] + positioning_tail_cap_k=None, + positioning_coin_gross_cap=None): # type: ignore[no-untyped-def] """POSITIONING single-sleeve Bybit track — thin wrapper on `build_bybit_sleeve_forward_nav` (keeps the capturability haircuts). See that function.""" return build_bybit_sleeve_forward_nav( repo, store, "positioning", today=today, at=at, universe=universe, cost_bps=cost_bps, - positioning_capacity_capital=positioning_capacity_capital, positioning_tail_cap_k=positioning_tail_cap_k) + positioning_capacity_capital=positioning_capacity_capital, positioning_tail_cap_k=positioning_tail_cap_k, + positioning_coin_gross_cap=positioning_coin_gross_cap) def build_bybit_paper_book(repo, store, *, capital, at, universe=None, unlock_events=None, @@ -865,7 +873,7 @@ def bybit_4edge_nav(context: AssetExecutionContext) -> dict: # type: ignore[typ universe = liquid_universe(store) or None # None → full universe (empty warehouse → empty book) # unlock calendar DB-first (operational unlock_events), JSON fallback — same source the live sleeve uses. unlock_events = load_unlock_events(operational_unlock_repo(s), f"{_data_dir()}/unlock_calendar.json") - from fxhnt.application.bybit_forward_track import POSITIONING_TAIL_CAP_K + from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP, POSITIONING_TAIL_CAP_K repo = ForwardNavRepo(s.operational_dsn) repo.migrate() today = _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%d") @@ -873,7 +881,8 @@ def bybit_4edge_nav(context: AssetExecutionContext) -> dict: # type: ignore[typ out = build_bybit_4edge_forward_nav( repo, store, today=today, at=at, universe=universe, cost_bps=s.bybit_cost_bps, unlock_events=unlock_events, - positioning_capacity_capital=s.paper_capital, positioning_tail_cap_k=POSITIONING_TAIL_CAP_K) + positioning_capacity_capital=s.paper_capital, positioning_tail_cap_k=POSITIONING_TAIL_CAP_K, + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) context.log.info( f"bybit_4edge_nav: T0={out['t0']}, {out['days']}d through {out['last_date']} " f"(+{out['total_return'] * 100:.2f}%, Sharpe {out['sharpe']:.2f}, " @@ -913,7 +922,7 @@ def bybit_4edge_levered_nav(context: AssetExecutionContext) -> dict: # type: ig try: universe = liquid_universe(store) or None unlock_events = load_unlock_events(operational_unlock_repo(s), f"{_data_dir()}/unlock_calendar.json") - from fxhnt.application.bybit_forward_track import POSITIONING_TAIL_CAP_K + from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP, POSITIONING_TAIL_CAP_K repo = ForwardNavRepo(s.operational_dsn) repo.migrate() today = _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%d") @@ -921,7 +930,8 @@ def bybit_4edge_levered_nav(context: AssetExecutionContext) -> dict: # type: ig out = build_bybit_4edge_levered_forward_nav( repo, store, today=today, at=at, universe=universe, cost_bps=s.bybit_cost_bps, unlock_events=unlock_events, - positioning_capacity_capital=s.paper_capital, positioning_tail_cap_k=POSITIONING_TAIL_CAP_K) + positioning_capacity_capital=s.paper_capital, positioning_tail_cap_k=POSITIONING_TAIL_CAP_K, + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) context.log.info( f"bybit_4edge_levered_nav: T0={out['t0']}, {out['days']}d through {out['last_date']} " f"(+{out['total_return'] * 100:.2f}%, Sharpe {out['sharpe']:.2f}, " @@ -954,7 +964,7 @@ def positioning_nav(context: AssetExecutionContext) -> dict: # type: ignore[typ store = TimescaleFeatureStore(s.operational_dsn, table="bybit_features") try: universe = liquid_universe(store) or None # None → full universe (empty warehouse → empty series) - from fxhnt.application.bybit_forward_track import POSITIONING_TAIL_CAP_K + from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP, POSITIONING_TAIL_CAP_K repo = ForwardNavRepo(s.operational_dsn) repo.migrate() today = _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%d") @@ -962,7 +972,8 @@ def positioning_nav(context: AssetExecutionContext) -> dict: # type: ignore[typ out = build_positioning_forward_nav( repo, store, today=today, at=at, universe=universe, cost_bps=s.bybit_cost_bps, - positioning_capacity_capital=s.paper_capital, positioning_tail_cap_k=POSITIONING_TAIL_CAP_K) + positioning_capacity_capital=s.paper_capital, positioning_tail_cap_k=POSITIONING_TAIL_CAP_K, + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) context.log.info( f"positioning_nav: T0={out['t0']}, {out['days']}d through {out['last_date']} " f"(+{out['total_return'] * 100:.2f}%, Sharpe {out['sharpe']:.2f}, " diff --git a/src/fxhnt/adapters/orchestration/migration_builders.py b/src/fxhnt/adapters/orchestration/migration_builders.py index 2fa1392..3af517d 100644 --- a/src/fxhnt/adapters/orchestration/migration_builders.py +++ b/src/fxhnt/adapters/orchestration/migration_builders.py @@ -103,7 +103,11 @@ def track_builders(settings: Settings, data_dir: str) -> TrackBuilders: # left open for the process lifetime of this one-time migration CLI. def _build_bybit_4edge() -> Any: from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore - from fxhnt.application.bybit_forward_track import POSITIONING_TAIL_CAP_K, BybitFourEdgeStrategy + from fxhnt.application.bybit_forward_track import ( + POSITIONING_COIN_GROSS_CAP, + POSITIONING_TAIL_CAP_K, + BybitFourEdgeStrategy, + ) from fxhnt.application.bybit_liquidity import liquid_universe from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo @@ -114,7 +118,8 @@ def track_builders(settings: Settings, data_dir: str) -> TrackBuilders: return BybitFourEdgeStrategy( store, universe=universe, cost_bps=settings.bybit_cost_bps, unlock_events=unlock_events, positioning_capacity_capital=settings.paper_capital, - positioning_tail_cap_k=POSITIONING_TAIL_CAP_K) + positioning_tail_cap_k=POSITIONING_TAIL_CAP_K, + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) builders["bybit_4edge"] = ("bybit_4edge_state", _build_bybit_4edge) @@ -124,6 +129,7 @@ def track_builders(settings: Settings, data_dir: str) -> TrackBuilders: from fxhnt.application.bybit_forward_track import ( _LEVERED_SLEEVE_SHAPE, MAX_BOOK_GROSS, + POSITIONING_COIN_GROSS_CAP, POSITIONING_TAIL_CAP_K, BybitFourEdgeStrategy, ) @@ -138,14 +144,19 @@ def track_builders(settings: Settings, data_dir: str) -> TrackBuilders: store, universe=universe, cost_bps=settings.bybit_cost_bps, unlock_events=unlock_events, leverage_shape=_LEVERED_SLEEVE_SHAPE, max_gross=MAX_BOOK_GROSS, positioning_capacity_capital=settings.paper_capital, - positioning_tail_cap_k=POSITIONING_TAIL_CAP_K) + positioning_tail_cap_k=POSITIONING_TAIL_CAP_K, + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) builders["bybit_4edge_levered"] = ("bybit_4edge_levered_state", _build_bybit_4edge_levered) # ---- positioning_nav: single-sleeve positioning edge on the liquid universe -------------------------- def _build_positioning() -> Any: from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore - from fxhnt.application.bybit_forward_track import POSITIONING_TAIL_CAP_K, BybitFourEdgeStrategy + from fxhnt.application.bybit_forward_track import ( + POSITIONING_COIN_GROSS_CAP, + POSITIONING_TAIL_CAP_K, + BybitFourEdgeStrategy, + ) from fxhnt.application.bybit_liquidity import liquid_universe store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features") @@ -153,7 +164,8 @@ def track_builders(settings: Settings, data_dir: str) -> TrackBuilders: return BybitFourEdgeStrategy( store, universe=universe, cost_bps=settings.bybit_cost_bps, sleeves=["positioning"], positioning_capacity_capital=settings.paper_capital, - positioning_tail_cap_k=POSITIONING_TAIL_CAP_K) + positioning_tail_cap_k=POSITIONING_TAIL_CAP_K, + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) builders["positioning"] = ("positioning_state", _build_positioning) diff --git a/src/fxhnt/application/bybit_forward_track.py b/src/fxhnt/application/bybit_forward_track.py index bfe0ce3..57b4703 100644 --- a/src/fxhnt/application/bybit_forward_track.py +++ b/src/fxhnt/application/bybit_forward_track.py @@ -140,6 +140,7 @@ class BybitFourEdgeStrategy: max_gross: float = MAX_BOOK_GROSS, positioning_capacity_capital: float | None = None, positioning_tail_cap_k: float | None = None, + positioning_coin_gross_cap: float | None = None, sleeve_returns: dict[str, dict[int, float]] | None = None) -> None: self._store = store self._universe = universe @@ -152,6 +153,7 @@ class BybitFourEdgeStrategy: # positioning-only capturability haircuts (gains-only capacity + tail-concentration cap); None = off. self._positioning_capacity_capital = positioning_capacity_capital self._positioning_tail_cap_k = positioning_tail_cap_k + self._positioning_coin_gross_cap = positioning_coin_gross_cap # PRECOMPUTED per-sleeve series (compute-once): when supplied the store reads are SKIPPED and these are # used directly — the persist Job computes the 4 sleeves once and reuses them for both references. self._sleeve_returns = sleeve_returns @@ -169,7 +171,8 @@ class BybitFourEdgeStrategy: self._store, sleeve, universe=self._universe, cost_bps=self._cost_bps, unlock_events=self._unlock_events, stablecoin_spot_panel=self._stablecoin_spot_panel, capacity_capital=self._positioning_capacity_capital, - tail_cap_k=self._positioning_tail_cap_k) + tail_cap_k=self._positioning_tail_cap_k, + coin_gross_cap=self._positioning_coin_gross_cap) if len(series) >= 2: # need >= 2 days to be a real sleeve in the book returns_by_sleeve[sleeve] = series diff --git a/src/fxhnt/application/bybit_paper_weights.py b/src/fxhnt/application/bybit_paper_weights.py index 684ff01..7534068 100644 --- a/src/fxhnt/application/bybit_paper_weights.py +++ b/src/fxhnt/application/bybit_paper_weights.py @@ -43,13 +43,17 @@ CARRY_SLEEVES = frozenset({"xsfunding"}) def sleeve_weights_and_contributions(store: Any, sleeve: str, *, universe: set[str] | None = None, unlock_events: list[dict[str, Any]] | None = None, + positioning_coin_gross_cap: float | None = None, ) -> dict[int, dict[str, tuple[float, float]]]: """`{epoch_day: {symbol: (weight, symbol_return)}}` for ONE Bybit edge, reusing the edge's EXACT engine. The sum `Σ_symbol weight·symbol_return` over a day equals that sleeve's GROSS (cost_bps=0) daily return from `sleeve_returns_from_store` — the reconciliation invariant. `universe` is the memory bound (liquid subset). `unlock_events` is required by the unlock sleeve (absent → empty, the edge is uncomputable). - READ-ONLY; writes nothing. `sleeve` accepts the sleeve name or the edge name.""" + `positioning_coin_gross_cap` (positioning sleeve ONLY; ignored by the others) must be passed the SAME + value used to compute the positioning sleeve's stored return, or the reconciliation invariant breaks — + the extracted weights would be uncapped while the stored return is capped. Default None (off) preserves + the byte-identical uncapped behaviour. READ-ONLY; writes nothing. `sleeve` accepts the sleeve or edge name.""" edge = _SLEEVE_TO_EDGE.get(sleeve, sleeve) if edge == "tstrend": from fxhnt.application.trend_runner import TrendRunner @@ -81,7 +85,8 @@ def sleeve_weights_and_contributions(store: Any, sleeve: str, *, universe: set[s from fxhnt.application.bybit_positioning_eval import positioning_weights_and_contributions # positioning does its OWN universe wrap (it also reads the long_ratio panel) — pass the raw store. - return positioning_weights_and_contributions(store, universe=universe) + return positioning_weights_and_contributions(store, universe=universe, + coin_gross_cap=positioning_coin_gross_cap) raise ValueError(f"unknown sleeve {sleeve!r} (known book sleeves: {_DEFAULT_BYBIT_SLEEVES})") diff --git a/src/fxhnt/application/bybit_positioning_eval.py b/src/fxhnt/application/bybit_positioning_eval.py index 7c3276b..31aa97f 100644 --- a/src/fxhnt/application/bybit_positioning_eval.py +++ b/src/fxhnt/application/bybit_positioning_eval.py @@ -262,6 +262,7 @@ def _book_breakdown(store: _FeatureStore, *, universe: set[str] | None, cost_bps def positioning_weights_and_contributions(store: _FeatureStore, *, universe: set[str] | None = None, direction: str = "contrarian", + coin_gross_cap: float | None = None, ) -> dict[int, dict[str, tuple[float, float]]]: """`{day: {symbol: (weight, symbol_return)}}` for the positioning edge, reusing the EXACT `positioning_weights` (demeaned, unit-gross) + `_next_day_returns` (causal d→d+1) the return engine @@ -277,7 +278,8 @@ def positioning_weights_and_contributions(store: _FeatureStore, *, universe: set close_panel = bounded.crypto_close_panel() if not close_panel: return {} - weights_by_day = positioning_weights(_ratio_by_day(long_ratio), direction=direction) + weights_by_day = positioning_weights(_ratio_by_day(long_ratio), direction=direction, + coin_gross_cap=coin_gross_cap) next_ret = _next_day_returns(close_panel) out: dict[int, dict[str, tuple[float, float]]] = {} for day in sorted(weights_by_day): diff --git a/tests/integration/test_bybit_paper_reconciliation.py b/tests/integration/test_bybit_paper_reconciliation.py index 13e3c6a..c1026f0 100644 --- a/tests/integration/test_bybit_paper_reconciliation.py +++ b/tests/integration/test_bybit_paper_reconciliation.py @@ -122,3 +122,40 @@ def test_unlock_without_calendar_is_empty() -> None: # No calendar → the edge is uncomputable → empty (never fabricated). assert sleeve_weights_and_contributions(store, "unlock", unlock_events=None) == {} store.close() + + +def _seed_positioning_shock(store: TimescaleFeatureStore, *, days: int = 4) -> None: + """One coin (LAB) is deep-over-short (ratio 0.05 -> huge demeaned weight, uncapped >> 0.07) so the + per-coin gross cap actually clips it; three coins sit near the mean (small weights, unaffected).""" + px = {"LABUSDT": 100.0, "AAAUSDT": 100.0, "BBBUSDT": 100.0, "CCCUSDT": 100.0} + for d in range(days): + 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"] *= 0.50 + + +def test_positioning_weights_reconcile_with_coin_gross_cap() -> None: + """SSOT invariant, CAPPED basis: when `positioning_coin_gross_cap` is set, the extracted per-symbol + weights must reconcile against the CAPPED stored return (not the uncapped one) — the paper-book + attribution must match whatever cap the live book actually used.""" + store = TimescaleFeatureStore("sqlite://", table="bybit_features") + _seed_positioning_shock(store) + extracted = sleeve_weights_and_contributions(store, "positioning", positioning_coin_gross_cap=0.07) + implied = reconciliation_series(extracted) + stored = sleeve_returns_from_store(store, "positioning", cost_bps=0.0, coin_gross_cap=0.07) + # and it must NOT trivially match the uncapped series (i.e. the cap is actually doing something — + # otherwise this test would pass vacuously even if the kwarg were silently dropped). + uncapped_stored = sleeve_returns_from_store(store, "positioning", cost_bps=0.0) + store.close() + assert stored, "positioning: engine produced no capped stored return series (seed too short?)" + common = set(implied) & set(stored) + assert common, "positioning: no common days between capped extracted weights and capped stored returns" + for day in sorted(stored): + assert day in implied, f"positioning day {day}: missing from capped extracted weights" + assert math.isclose(implied[day], stored[day], rel_tol=1e-9, abs_tol=1e-12), ( + f"positioning day {day}: Σ w·r = {implied[day]!r} != capped stored {stored[day]!r}") + assert any(not math.isclose(stored[d], uncapped_stored[d], rel_tol=1e-9, abs_tol=1e-12) + for d in common), "cap had no effect on the fixture — strengthen the seed to exercise it" diff --git a/tests/integration/test_positioning_forward_track.py b/tests/integration/test_positioning_forward_track.py index 07644d0..a82d8ec 100644 --- a/tests/integration/test_positioning_forward_track.py +++ b/tests/integration/test_positioning_forward_track.py @@ -141,3 +141,30 @@ def test_driver_is_read_only(tmp_path) -> None: store.close() assert out["t0"] # produced a real result assert repo.nav_history("positioning") # persisted to the DB (allowed) + + +def test_forward_nav_applies_coin_gross_cap() -> None: + """The forward NAV (BybitFourEdgeStrategy store-read path) must apply the per-coin cap when given, so it + reconciles against the capped ref instead of diverging. Shock store: one over-short coin craters ~50%.""" + from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore + from fxhnt.application.bybit_forward_track import BybitFourEdgeStrategy + + _DAY = 86_400 + store = TimescaleFeatureStore("sqlite://", table="bybit_features") + 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"] *= 0.50 + uncapped_rows, _ = BybitFourEdgeStrategy(store, sleeves=["positioning"]).advance(None, {}) + capped_rows, _ = BybitFourEdgeStrategy( + store, sleeves=["positioning"], positioning_coin_gross_cap=0.07).advance(None, {}) + store.close() + assert uncapped_rows and capped_rows, "forward strategy produced no rows — fixture broken" + u = dict(uncapped_rows) + c = dict(capped_rows) + worst = min(u, key=lambda k: u[k]) + assert c[worst] > u[worst], f"cap must reduce the forward worst day: uncapped={u[worst]}, capped={c[worst]}" From fb4cb49b18365e2ac69288bc527c08515a9857ee Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 19:46:21 +0000 Subject: [PATCH 11/15] fix(positioning): cap live paper-book positions + report-ref path (coin_gross_cap; Task-7-review Critical+Important) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Threads positioning_coin_gross_cap through the two remaining uncapped paths found by the Task-7 whole-branch review: - Critical: combined_symbol_weights / combined_symbol_weights_and_returns / latest_raw_sleeve_weights / derive_and_persist_bybit_paper_book / build_bybit_paper_book now accept positioning_coin_gross_cap and forward it into sleeve_weights_and_contributions, so the live Bybit paper-book's ACTUAL persisted positions honor the same per-coin cap as the reconciled return. The nightly bybit_paper_book asset now passes POSITIONING_COIN_GROSS_CAP. - Important: bybit_4edge_backtest_summary gained a positioning_coin_gross_cap sibling next to positioning_tail_cap_k, forwarded into BybitFourEdgeStrategy; report_backtest_summary now passes POSITIONING_COIN_GROSS_CAP alongside POSITIONING_TAIL_CAP_K. Scope: coin_gross_cap only (the EX-ANTE weight cap) — capacity_capital and tail_cap_k are gains-only return haircuts and do not belong on the weight path. Default None everywhere (OFF = byte-identical). Co-Authored-By: Claude Opus 4.8 --- src/fxhnt/adapters/orchestration/assets.py | 15 ++++++-- src/fxhnt/application/backtest_refs_report.py | 12 +++--- src/fxhnt/application/bybit_forward_track.py | 10 +++-- src/fxhnt/application/bybit_paper_book.py | 37 ++++++++++++++----- .../test_bybit_paper_reconciliation.py | 23 ++++++++++++ 5 files changed, 76 insertions(+), 21 deletions(-) diff --git a/src/fxhnt/adapters/orchestration/assets.py b/src/fxhnt/adapters/orchestration/assets.py index a11fc7c..730ca16 100644 --- a/src/fxhnt/adapters/orchestration/assets.py +++ b/src/fxhnt/adapters/orchestration/assets.py @@ -178,7 +178,7 @@ def build_positioning_forward_nav(repo, store, *, today, at, universe=None, cost def build_bybit_paper_book(repo, store, *, capital, at, universe=None, unlock_events=None, - enabled=True): # type: ignore[no-untyped-def] + enabled=True, positioning_coin_gross_cap=None): # type: ignore[no-untyped-def] """Plain (Dagster-free, unit-testable) driver for the Bybit LIVE paper book: derive the naive-eq-wt 4-edge book's per-symbol positions on the latest Bybit prices and persist them (positions/trades/nav) via `derive_and_persist_bybit_paper_book` into `repo` (a `PaperRepo` — bybit is its sole venue). @@ -187,14 +187,19 @@ def build_bybit_paper_book(repo, store, *, capital, at, universe=None, unlock_ev Charges the per-COIN MEASURED cost (real L1 spread from `bybit_real_spread`, taker_fee + 0.5·spread) so the forward `/paper` step ties to the cockpit Backtest NET; falls back to the flat 5.5bp when no spread refresh - has run.""" + has run. + + `positioning_coin_gross_cap` (positioning sleeve ONLY) forwards into `derive_and_persist_bybit_paper_book` + so the LIVE persisted positions honor the same per-coin cap the reconciled return uses. Default None + preserves the byte-identical uncapped behaviour; the nightly asset passes `POSITIONING_COIN_GROSS_CAP`.""" from fxhnt.application.bybit_paper_book import derive_and_persist_bybit_paper_book from fxhnt.application.bybit_spread_refresh import read_real_spread_by_coin real_spread = read_real_spread_by_coin(repo) or None n = derive_and_persist_bybit_paper_book( repo, store, capital=capital, at=at, universe=universe, - unlock_events=unlock_events, enabled=enabled, real_spread_by_coin=real_spread) + unlock_events=unlock_events, enabled=enabled, real_spread_by_coin=real_spread, + positioning_coin_gross_cap=positioning_coin_gross_cap) # PROACTIVE: re-materialize the headline nav-summary now the bybit nav changed, so the first / + /paper # request is already fast (read 1 row, not the full nav). The read path self-heals if this is skipped, so # a refresh failure must NEVER fail the book write — guarded. @@ -1088,6 +1093,7 @@ def bybit_paper_book(context: AssetExecutionContext) -> dict: # type: ignore[ty from fxhnt.adapters.persistence.paper_repo import PaperRepo from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore + from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP from fxhnt.application.bybit_liquidity import liquid_universe from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo from fxhnt.config import get_settings @@ -1104,7 +1110,8 @@ def bybit_paper_book(context: AssetExecutionContext) -> dict: # type: ignore[ty unlock_events = load_unlock_events(operational_unlock_repo(s), f"{_data_dir()}/unlock_calendar.json") n = build_bybit_paper_book( repo, store, capital=s.paper_capital, at=dt.datetime.now(dt.UTC), - universe=universe, unlock_events=unlock_events, enabled=True) + universe=universe, unlock_events=unlock_events, enabled=True, + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) context.log.info(f"bybit_paper_book: persisted {n} open positions (venue=bybit)") return {"positions": n, "enabled": True} except (urllib.error.URLError, TimeoutError, ConnectionError, OSError) as e: # transient: don't kill the run diff --git a/src/fxhnt/application/backtest_refs_report.py b/src/fxhnt/application/backtest_refs_report.py index 8b47088..9fe35f2 100644 --- a/src/fxhnt/application/backtest_refs_report.py +++ b/src/fxhnt/application/backtest_refs_report.py @@ -2,6 +2,7 @@ import logging import os from fxhnt.application.bybit_forward_track import ( + POSITIONING_COIN_GROSS_CAP, POSITIONING_TAIL_CAP_K, bybit_4edge_backtest_summary, bybit_report_sid_kwargs, @@ -25,10 +26,10 @@ _REPORT_SIDS = ("bybit_4edge", "bybit_4edge_levered", "positioning") def report_backtest_summary(strategy_id, store): # -> BacktestSummary | None """Backtest ref for a `report`-kind bybit strategy: dispatch to the real book engine `bybit_4edge_backtest_summary` with the per-sid kwargs. Reconstructs the liquid universe, unlock - calendar, and positioning capturability haircuts internally (from the store + settings) so the ref - reconciles against the SAME capturable book the forward track records. cost_bps=5.5 matches the live - BybitFourEdge basis and the `sleeve` kind. Built from the store (sleeve_returns=None) so the - positioning haircut kwargs are applied by the engine.""" + calendar, and positioning capturability haircuts + per-coin gross cap internally (from the store + + settings) so the ref reconciles against the SAME capturable book the forward track records. cost_bps=5.5 + matches the live BybitFourEdge basis and the `sleeve` kind. Built from the store (sleeve_returns=None) so + the positioning haircut + coin_gross_cap kwargs are applied by the engine.""" if strategy_id not in _REPORT_SIDS: raise ValueError(f"report_backtest_summary: {strategy_id!r} is not a report-kind sid {_REPORT_SIDS}") settings = get_settings() @@ -42,7 +43,8 @@ def report_backtest_summary(strategy_id, store): # -> BacktestSummary | None common = dict( universe=universe, cost_bps=5.5, positioning_capacity_capital=settings.paper_capital, - positioning_tail_cap_k=POSITIONING_TAIL_CAP_K) + positioning_tail_cap_k=POSITIONING_TAIL_CAP_K, + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) # unlock_events is always passed: the positioning sleeve construction (both here and in # `_persist_bybit_book`) ignores it entirely, so unifying to always-pass is a no-op for that sid — see # `bybit_report_sid_kwargs`. diff --git a/src/fxhnt/application/bybit_forward_track.py b/src/fxhnt/application/bybit_forward_track.py index 57b4703..89ab113 100644 --- a/src/fxhnt/application/bybit_forward_track.py +++ b/src/fxhnt/application/bybit_forward_track.py @@ -216,6 +216,7 @@ def bybit_4edge_backtest_summary(store: _FeatureStore, *, universe: set[str] | N strategy_id: str = "bybit_4edge", positioning_capacity_capital: float | None = None, positioning_tail_cap_k: float | None = None, + positioning_coin_gross_cap: float | None = None, sleeve_returns: dict[str, dict[int, float]] | None = None): # type: ignore[no-untyped-def] """Backtest reference for the bybit_4edge RECONCILIATION gate: the full-history metrics of the SAME naive eq-wt funding-net 4-edge book the forward track records (`BybitFourEdgeStrategy`). Persisted under @@ -224,15 +225,18 @@ def bybit_4edge_backtest_summary(store: _FeatureStore, *, universe: set[str] | N (a false-PASS risk on a real-capital gate). cagr/vol/sharpe/maxDD from the full series; the gauntlet verdict (dsr/is/oos/pvalue) from a 60/40 IS/OOS split, mirroring the equity-factor backtest_summary rows. - `positioning_capacity_capital`/`positioning_tail_cap_k` apply the positioning capturability haircuts (only - when this builds from the store, i.e. `sleeve_returns` is None; a precomputed `sleeve_returns` must already - have them baked in — the persist Job computes the positioning series with them).""" + `positioning_capacity_capital`/`positioning_tail_cap_k` apply the positioning capturability haircuts; + `positioning_coin_gross_cap` applies the per-coin EX-ANTE weight cap (shapes the position, not a return + haircut) — all three only take effect when this builds from the store, i.e. `sleeve_returns` is None; a + precomputed `sleeve_returns` must already have them baked in — the persist Job computes the positioning + series with them).""" rows, _ = BybitFourEdgeStrategy( store, universe=universe, sleeves=sleeves, cost_bps=cost_bps, unlock_events=unlock_events, stablecoin_spot_panel=stablecoin_spot_panel, leverage_shape=leverage_shape, max_gross=max_gross, positioning_capacity_capital=positioning_capacity_capital, positioning_tail_cap_k=positioning_tail_cap_k, + positioning_coin_gross_cap=positioning_coin_gross_cap, sleeve_returns=sleeve_returns).advance(None, {}) return backtest_summary_from_rows(strategy_id, rows) diff --git a/src/fxhnt/application/bybit_paper_book.py b/src/fxhnt/application/bybit_paper_book.py index 737b289..3bc4b40 100644 --- a/src/fxhnt/application/bybit_paper_book.py +++ b/src/fxhnt/application/bybit_paper_book.py @@ -49,6 +49,7 @@ def combined_symbol_weights(store: Any, *, universe: set[str] | None = None, unlock_events: list[dict[str, Any]] | None = None, sleeves: tuple[str, ...] = _DEFAULT_BYBIT_SLEEVES, as_of_day: int | None = None, + positioning_coin_gross_cap: float | None = None, ) -> tuple[dict[str, dict[str, float]], int | None]: """`({sleeve: {symbol: combined_weight}}, as_of_day)` for the LATEST day of each sleeve, combined NAIVE EQUAL-WEIGHT (1/n per sleeve). `combined_weight = (1/n)·sleeve_weight` where `sleeve_weight` is the @@ -57,7 +58,9 @@ def combined_symbol_weights(store: Any, *, universe: set[str] | None = None, Each sleeve's weights are taken on ITS OWN latest available day (the engines emit different last days); the returned `as_of_day` is the max latest day across the sleeves (the book's run_date). A sleeve with no weights (no data / uncomputable, e.g. unlock with no calendar) is simply absent. `as_of_day` (epoch day) - can be pinned for a deterministic backfill day. Pure given the store; READ-ONLY.""" + can be pinned for a deterministic backfill day. `positioning_coin_gross_cap` (positioning sleeve ONLY) is + forwarded straight into `sleeve_weights_and_contributions` — default None preserves the byte-identical + uncapped behaviour. Pure given the store; READ-ONLY.""" n = len(sleeves) if n == 0: return {}, None @@ -65,7 +68,8 @@ def combined_symbol_weights(store: Any, *, universe: set[str] | None = None, latest_day: int | None = None for sleeve in sleeves: by_day = sleeve_weights_and_contributions(store, sleeve, universe=universe, - unlock_events=unlock_events) + unlock_events=unlock_events, + positioning_coin_gross_cap=positioning_coin_gross_cap) days = [d for d in by_day if as_of_day is None or d <= as_of_day] if not days: continue @@ -79,7 +83,8 @@ def combined_symbol_weights(store: Any, *, universe: set[str] | None = None, def latest_raw_sleeve_weights(store: Any, *, universe: set[str] | None = None, unlock_events: list[dict[str, Any]] | None = None, sleeves: tuple[str, ...] = _DEFAULT_BYBIT_SLEEVES, - as_of_day: int | None = None) -> dict[str, float]: + as_of_day: int | None = None, + positioning_coin_gross_cap: float | None = None) -> dict[str, float]: """`{symbol: raw_target_weight}` for the Bybit 4-edge book at its LATEST as_of, BEFORE any risk overlay — the source the real-paper / testnet execution leg places orders against. @@ -92,9 +97,11 @@ def latest_raw_sleeve_weights(store: Any, *, universe: set[str] | None = None, Pure + READ-ONLY against the warehouse `store` (writes nothing); memory-bounded on the liquid `universe`. Net-zero symbols (a long sleeve cancelled by a short sleeve) are dropped. `as_of_day` (epoch day) pins a deterministic day; default = each sleeve's own latest. The companion as_of day is available via - `combined_symbol_weights` when the caller needs it (e.g. an idempotency marker).""" + `combined_symbol_weights` when the caller needs it (e.g. an idempotency marker). `positioning_coin_gross_cap` + forwards into `combined_symbol_weights` — default None preserves the byte-identical uncapped behaviour.""" combined, _day = combined_symbol_weights( - store, universe=universe, unlock_events=unlock_events, sleeves=sleeves, as_of_day=as_of_day) + store, universe=universe, unlock_events=unlock_events, sleeves=sleeves, as_of_day=as_of_day, + positioning_coin_gross_cap=positioning_coin_gross_cap) out: dict[str, float] = {} for by_symbol in combined.values(): for sym, w in by_symbol.items(): @@ -107,12 +114,16 @@ def combined_symbol_weights_and_returns( unlock_events: list[dict[str, Any]] | None = None, sleeves: tuple[str, ...] = _DEFAULT_BYBIT_SLEEVES, as_of_day: int | None = None, + positioning_coin_gross_cap: float | None = None, ) -> tuple[dict[str, dict[str, float]], dict[str, dict[str, float]], int | None]: """`(combined_weights, returns, as_of_day)` — like `combined_symbol_weights` but ALSO surfaces each sleeve's TRUE per-symbol return on its OWN latest (bundled) day: `returns[sleeve][symbol]` is the `r` in the `(weight, return)` pair (carry = basis + FUNDING, unlock = short_leg_pnl incl. hedge + squeeze, trend/positioning = price). Fed to `persist_paper_book`'s return-defined book so the live forward stepper - books the TRUE edge return (no naked perp price-MTM, no dropped funding/hedge). Pure; READ-ONLY.""" + books the TRUE edge return (no naked perp price-MTM, no dropped funding/hedge). `positioning_coin_gross_cap` + (positioning sleeve ONLY) forwards into `sleeve_weights_and_contributions` so the LIVE paper-book positions + honor the same per-coin cap the reconciled return uses — default None preserves the byte-identical + uncapped behaviour. Pure; READ-ONLY.""" n = len(sleeves) if n == 0: return {}, {}, None @@ -121,7 +132,8 @@ def combined_symbol_weights_and_returns( latest_day: int | None = None for sleeve in sleeves: by_day = sleeve_weights_and_contributions(store, sleeve, universe=universe, - unlock_events=unlock_events) + unlock_events=unlock_events, + positioning_coin_gross_cap=positioning_coin_gross_cap) days = [d for d in by_day if as_of_day is None or d <= as_of_day] if not days: continue @@ -145,7 +157,8 @@ def derive_and_persist_bybit_paper_book(repo: PaperRepo, store: Any, *, capital: enabled: bool = True, cost_rate: float = _BYBIT_COST_RATE, as_of_day: int | None = None, - real_spread_by_coin: dict[str, float] | None = None) -> int: + real_spread_by_coin: dict[str, float] | None = None, + positioning_coin_gross_cap: float | None = None) -> int: """Derive the Bybit live paper book for its latest day and persist it through the shared `persist_paper_book` seam into `repo` (bybit is `PaperRepo`'s sole venue). Returns the number of open positions persisted (0 when disabled / no data). @@ -167,11 +180,17 @@ def derive_and_persist_bybit_paper_book(repo: PaperRepo, store: Any, *, capital: SAME taker_fee + 0.5·real_spread(coin) the cockpit Backtest charges), so the forward `/paper` step ties to the Backtest NET; else the flat Bybit 5.5bp turnover haircut (the pre-real-spread path). + `positioning_coin_gross_cap` (positioning sleeve ONLY) forwards into `combined_symbol_weights_and_returns` + so the LIVE persisted positions honor the same per-coin cap the reconciled return uses — a single + over-short coin can no longer take an outsized real position. Default None preserves the byte-identical + uncapped behaviour. + READ-ONLY against the warehouse `store`; only the paper store is written. `enabled=False` is a no-op.""" if not enabled: return 0 combined, ret_map, day = combined_symbol_weights_and_returns( - store, universe=universe, unlock_events=unlock_events, as_of_day=as_of_day) + store, universe=universe, unlock_events=unlock_events, as_of_day=as_of_day, + positioning_coin_gross_cap=positioning_coin_gross_cap) if not combined or day is None: return 0 run_date = _epoch_day_to_iso(day) diff --git a/tests/integration/test_bybit_paper_reconciliation.py b/tests/integration/test_bybit_paper_reconciliation.py index c1026f0..e9ad9a5 100644 --- a/tests/integration/test_bybit_paper_reconciliation.py +++ b/tests/integration/test_bybit_paper_reconciliation.py @@ -159,3 +159,26 @@ def test_positioning_weights_reconcile_with_coin_gross_cap() -> None: f"positioning day {day}: Σ w·r = {implied[day]!r} != capped stored {stored[day]!r}") assert any(not math.isclose(stored[d], uncapped_stored[d], rel_tol=1e-9, abs_tol=1e-12) for d in common), "cap had no effect on the fixture — strengthen the seed to exercise it" + + +def test_paper_book_positions_capped_by_coin_gross_cap() -> None: + """The live paper-book per-symbol weights must honor the per-coin cap so executed positions match the + capped reconciled return (SSOT). On a store with one dominant positioning coin, the capped weight for that + coin is strictly smaller than uncapped.""" + from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore + from fxhnt.application.bybit_paper_book import combined_symbol_weights_and_returns + _DAY = 86_400 + store = TimescaleFeatureStore("sqlite://", table="bybit_features") + for d in range(4): + store.write_features("LABUSDT", [(d * _DAY, {"long_ratio": 0.02, "close": 100.0})]) + for i, sym in enumerate(("AAAUSDT", "BBBUSDT", "CCCUSDT", "DDDUSDT")): + store.write_features(sym, [(d * _DAY, {"long_ratio": 0.45 + 0.02 * i, "close": 100.0})]) + w_unc, _r, _d = combined_symbol_weights_and_returns(store, sleeves=("positioning",)) + w_cap, _r2, _d2 = combined_symbol_weights_and_returns( + store, sleeves=("positioning",), positioning_coin_gross_cap=0.07) + store.close() + assert w_unc.get("positioning") and w_cap.get("positioning"), "no positioning weights — fixture broken" + lab_unc = abs(w_unc["positioning"].get("LABUSDT", 0.0)) + lab_cap = abs(w_cap["positioning"].get("LABUSDT", 0.0)) + assert lab_unc > 0.0 + assert lab_cap < lab_unc, f"cap must shrink the dominant coin's position: unc={lab_unc}, cap={lab_cap}" From dd527e66589a56fde420f4bfb1c3f3e55e1b5edd Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 19:53:08 +0000 Subject: [PATCH 12/15] fix(positioning): thread coin_gross_cap into bybit-paper-book CLI seed path The review of fb4cb49 found one more silently-uncapped call into build_bybit_paper_book: the `bybit-paper-book` CLI command (cli.py) used to manually seed/re-seed the SAME live Bybit paper book the nightly Dagster asset populates. It mirrored the nightly asset's pre-fix call exactly and was missed because it wasn't in that task's file list. Pass positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP, matching the nightly bybit_paper_book asset call site. Default-None behavior elsewhere is unaffected. --- src/fxhnt/cli.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/fxhnt/cli.py b/src/fxhnt/cli.py index 48c2d75..b6f9ba7 100644 --- a/src/fxhnt/cli.py +++ b/src/fxhnt/cli.py @@ -3281,6 +3281,7 @@ def bybit_paper_book_cmd( from fxhnt.adapters.orchestration.assets import _data_dir, build_bybit_paper_book from fxhnt.adapters.persistence.paper_repo import PaperRepo from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore + from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP from fxhnt.application.bybit_liquidity import liquid_universe from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo @@ -3299,7 +3300,8 @@ def bybit_paper_book_cmd( repo.migrate() n = build_bybit_paper_book( repo, store, capital=settings.paper_capital, at=dt.datetime.now(dt.UTC), - universe=universe, unlock_events=unlock_events, enabled=True) + universe=universe, unlock_events=unlock_events, enabled=True, + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) finally: store.close() From f325a3707096996023e8a89f5a04afaa5310c86e Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 20:00:07 +0000 Subject: [PATCH 13/15] fix(positioning): cap testnet exec-sizing path (4th uncapped money-path found by exhaustive sweep) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An exhaustive grep of every build_bybit_paper_book / combined_symbol_weights* / latest_raw_sleeve_weights caller found bybit_testnet_run.py sized the executed testnet positions UNCAPPED (latest_raw_sleeve_weights + combined_symbol_weights called without positioning_coin_gross_cap). The cron is suspended (no creds) but it is a real order-placing path — the moment it arms it would size concentrated positions. Thread POSITIONING_COIN_GROSS_CAP the same as every other exec path. 35 testnet tests green, import + lint clean (no circular import). Co-Authored-By: Claude Opus 4.8 --- src/fxhnt/application/bybit_testnet_run.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/fxhnt/application/bybit_testnet_run.py b/src/fxhnt/application/bybit_testnet_run.py index 3979d9c..ae63f15 100644 --- a/src/fxhnt/application/bybit_testnet_run.py +++ b/src/fxhnt/application/bybit_testnet_run.py @@ -17,6 +17,7 @@ from typing import Any # `latest_raw_sleeve_weights` / `combined_symbol_weights` are imported at MODULE level (not locally like the # other body deps) so tests can `monkeypatch.setattr(run_mod, "latest_raw_sleeve_weights", ...)`. +from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP from fxhnt.application.bybit_paper_book import combined_symbol_weights, latest_raw_sleeve_weights @@ -112,9 +113,14 @@ def run_bybit_testnet(settings: Any, *, paper_envelope: float, execute: bool, try: universe = _testnet_universe(store) unlock_events = _testnet_unlock_events(s) - weights = latest_raw_sleeve_weights(store, universe=universe, unlock_events=unlock_events) + # Cap the executed testnet positions the SAME way the modeled/forward/paper-book paths do, so a single + # over-short coin can't take an outsized real position (the reconciled return is capped; the executed + # position must match). POSITIONING_COIN_GROSS_CAP is the ex-ante per-coin weight cap (0.07). + weights = latest_raw_sleeve_weights(store, universe=universe, unlock_events=unlock_events, + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) _combined, as_of_day = combined_symbol_weights( - store, universe=universe, unlock_events=unlock_events) + store, universe=universe, unlock_events=unlock_events, + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) marks = _testnet_marks(store, universe=universe) finally: store.close() From 3a26b87f15e5173b860bcfd0ff7732ca79420204 Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 20:17:20 +0000 Subject: [PATCH 14/15] fix(positioning): cap the last 2 live paths (paper-backfill + allocation) + coverage-audit regression test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An exhaustive sweep of every caller in the positioning-cap dependency chain found two more live paths still uncapped: the Bybit paper-book BACKFILL (bybit_paper_backfill.py, sizes/persists live paper positions across history) and the honest-reference allocation inputs (allocation_honest_inputs.py, feeds record_allocation, the LIVE strategy allocation engine's ex-ante capital sizing). Threads positioning_coin_gross_cap (opt-in, default None) through both, matching the existing pattern: - bybit_paper_backfill.py: _extract_per_sleeve + backfill_bybit_paper_book now accept and forward the cap; cli.py's bybit-paper-backfill command passes POSITIONING_COIN_GROSS_CAP. - allocation_honest_inputs.py: _sleeve_inputs + _book_curve + _deploy_curve + honest_allocation_inputs now accept and forward the cap; allocation_ingest.py's record_allocation (called from the cockpit_forward Dagster asset) passes POSITIONING_COIN_GROSS_CAP into honest_allocation_inputs. Adds tests/unit/test_positioning_cap_coverage_audit.py — a source-level regression guard asserting every LIVE-book/position/allocation callsite passes positioning_coin_gross_cap. Fixed a boundary bug in the brief's own _func_body helper regex (bare ^\S with re.MULTILINE always matches position 0 of the sliced remainder, truncating every function body to its signature line) so the audit actually inspects each function's body instead of false-failing on already-capped callsites. Research/eval callers (verify_positioning_edge, walk_forward_positioning, look_ahead_audit, _variant_weights, _drop_top_n, _liquidity_sweep, positioning_metalabel, bybit_overlay_ab, vrp_eval, onchain_fundamental_eval, spread_overlay, honest_report) are untouched — they stay uncapped by design. Co-Authored-By: Claude Opus 4.8 --- .../application/allocation_honest_inputs.py | 40 +++++++++++++------ src/fxhnt/application/allocation_ingest.py | 4 +- src/fxhnt/application/bybit_paper_backfill.py | 16 ++++++-- src/fxhnt/cli.py | 4 +- .../test_positioning_cap_coverage_audit.py | 40 +++++++++++++++++++ 5 files changed, 87 insertions(+), 17 deletions(-) create mode 100644 tests/unit/test_positioning_cap_coverage_audit.py diff --git a/src/fxhnt/application/allocation_honest_inputs.py b/src/fxhnt/application/allocation_honest_inputs.py index 8364b7c..e8e97ec 100644 --- a/src/fxhnt/application/allocation_honest_inputs.py +++ b/src/fxhnt/application/allocation_honest_inputs.py @@ -68,11 +68,15 @@ def _iso(epoch_day: int) -> str: def _sleeve_inputs(store: Any, sleeve: str, - unlock_events: list[dict[str, Any]] | None) -> tuple[dict, dict] | None: + unlock_events: list[dict[str, Any]] | None, + positioning_coin_gross_cap: float | None = None) -> tuple[dict, dict] | None: """AUM-INDEPENDENT heavy inputs for a sleeve: (weights_by_day, turnover_panel). Load ONCE, reprice across AUMs via honest_sleeve_returns/capacity_curve — the turnover panel read is the memory cost, so it must NOT - be reloaded per AUM. None when the sleeve is uncomputable (e.g. unlock with no events). READ-ONLY.""" - wbd = sleeve_weights_and_contributions(store, sleeve, unlock_events=unlock_events) + be reloaded per AUM. None when the sleeve is uncomputable (e.g. unlock with no events). + `positioning_coin_gross_cap` (positioning sleeve ONLY) forwards into `sleeve_weights_and_contributions` — + default None preserves the byte-identical uncapped behaviour. READ-ONLY.""" + wbd = sleeve_weights_and_contributions(store, sleeve, unlock_events=unlock_events, + positioning_coin_gross_cap=positioning_coin_gross_cap) if not wbd: return None symbols = sorted({s for row in wbd.values() for s in row}) @@ -81,11 +85,14 @@ def _sleeve_inputs(store: Any, sleeve: str, def _book_curve(store: Any, spread_bps: dict[str, float], unlock_events: list[dict[str, Any]] | None, - aums, *, participation_cap: float, impact_k: float) -> dict[float, dict[int, float]]: + aums, *, participation_cap: float, impact_k: float, + positioning_coin_gross_cap: float | None = None) -> dict[float, dict[int, float]]: """{aum: risk-parity BOOK series} across `aums`, loading each `_BOOK_SLEEVES` sleeve's inputs ONCE then - repricing per AUM. Empty sleeves (uncomputable) are dropped; an AUM with no computable sleeve -> {}.""" + repricing per AUM. Empty sleeves (uncomputable) are dropped; an AUM with no computable sleeve -> {}. + `positioning_coin_gross_cap` forwards into `_sleeve_inputs` (positioning sleeve ONLY).""" loaded = {sleeve: inp for sleeve in _BOOK_SLEEVES - if (inp := _sleeve_inputs(store, sleeve, unlock_events)) is not None} + if (inp := _sleeve_inputs(store, sleeve, unlock_events, + positioning_coin_gross_cap=positioning_coin_gross_cap)) is not None} out: dict[float, dict[int, float]] = {} for aum in aums: per: dict[str, dict[int, float]] = {} @@ -100,14 +107,18 @@ def _book_curve(store: Any, spread_bps: dict[str, float], unlock_events: list[di def _deploy_curve(store: Any, sid: str, spread_bps: dict[str, float], unlock_events: list[dict[str, Any]] | None, aums, *, - participation_cap: float, impact_k: float) -> dict[float, dict[int, float]]: + participation_cap: float, impact_k: float, + positioning_coin_gross_cap: float | None = None) -> dict[float, dict[int, float]]: """{aum: honest series} for a crypto deploy `sid` across `aums`, loading AUM-independent inputs ONCE. - `bybit_4edge` -> the risk-parity book; a sleeve sid -> its own series (reuses `capacity_curve`); other -> {}.""" + `bybit_4edge` -> the risk-parity book; a sleeve sid -> its own series (reuses `capacity_curve`); other -> {}. + `positioning_coin_gross_cap` forwards through (positioning sleeve ONLY).""" if sid == "bybit_4edge": return _book_curve(store, spread_bps, unlock_events, aums, - participation_cap=participation_cap, impact_k=impact_k) + participation_cap=participation_cap, impact_k=impact_k, + positioning_coin_gross_cap=positioning_coin_gross_cap) if sid in _CRYPTO_DEPLOY_SIDS: - inputs = _sleeve_inputs(store, sid, unlock_events) + inputs = _sleeve_inputs(store, sid, unlock_events, + positioning_coin_gross_cap=positioning_coin_gross_cap) if inputs is None: return {aum: {} for aum in aums} wbd, turnover = inputs @@ -120,6 +131,7 @@ def honest_allocation_inputs( store: Any, spread_bps: dict[str, float], unlock_events: list[dict[str, Any]] | None, *, deploy_sids, target_aum: float, aums, participation_cap: float = 0.03, impact_k: float = 1.0, min_sharpe: float = 1.0, equity_default_ceiling: float = 10_000_000.0, + positioning_coin_gross_cap: float | None = None, ) -> tuple[dict[str, dict[str, float]], dict[str, float | None]]: """`(returns_by_strategy, capacity_ceiling_by_strategy)` for the deploy set — the allocation engine's honest inputs. @@ -127,7 +139,10 @@ def honest_allocation_inputs( For each crypto deploy sid: its honest series at `target_aum` (returned STRING-keyed, see the module docstring seam) plus a capacity ceiling = `capacity_ceiling_from_curve({aum: sharpe_of(series(aum)) for aum in aums}, min_sharpe)`. Non-crypto sids get `equity_default_ceiling` and NO series (the caller - supplies their reference). PURE + read-only.""" + supplies their reference). `positioning_coin_gross_cap` (positioning sleeve ONLY) forwards into + `_deploy_curve` / `_sleeve_inputs` so the honest-reference basis the LIVE allocation engine + (`record_allocation`) sizes against honors the same per-coin cap the live capped return uses — default + None preserves the byte-identical uncapped behaviour. PURE + read-only.""" returns_by_strategy: dict[str, dict[str, float]] = {} ceilings: dict[str, float | None] = {} for sid in deploy_sids: @@ -137,7 +152,8 @@ def honest_allocation_inputs( # AUM-independent panels loaded ONCE inside _deploy_curve; base + curve are repriced from them. all_aums = sorted({float(target_aum), *(float(a) for a in aums)}) curve_series = _deploy_curve(store, sid, spread_bps, unlock_events, all_aums, - participation_cap=participation_cap, impact_k=impact_k) + participation_cap=participation_cap, impact_k=impact_k, + positioning_coin_gross_cap=positioning_coin_gross_cap) base = curve_series.get(float(target_aum), {}) if base: # int epoch-day -> ISO date string, the SAME key the nav fallback uses, consistently across every diff --git a/src/fxhnt/application/allocation_ingest.py b/src/fxhnt/application/allocation_ingest.py index d71fb4c..e1a2ac0 100644 --- a/src/fxhnt/application/allocation_ingest.py +++ b/src/fxhnt/application/allocation_ingest.py @@ -22,6 +22,7 @@ from fxhnt.application.allocation_policy import ( allocation_policy_hash, resolve_allocation_policy, ) +from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP from fxhnt.application.forward_ingest import CockpitBacktestRefProvider from fxhnt.application.funding_status import FundingStatus, compute_funding from fxhnt.application.gate_policy import POLICY, POLICY_VERSION, policy_hash @@ -89,7 +90,8 @@ def record_allocation(dsn: str, equity: float, *, store=None, spread_bps=None, deploy_sids=funded_sids, target_aum=equity, aums=[35_000, 100_000, 350_000, 1_000_000, 10_000_000], min_sharpe=getattr(pol, "capacity_min_sharpe", 1.0), - equity_default_ceiling=getattr(pol, "equity_default_ceiling", 10_000_000.0)) + equity_default_ceiling=getattr(pol, "equity_default_ceiling", 10_000_000.0), + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) # Equity leg (sub-project: equity leg deployable, Task 3): the ibkr-equity deploy sids (today only # `multistrat`) get their OWN cost-netted sizing series + real-ADV capacity ceiling here, merged on diff --git a/src/fxhnt/application/bybit_paper_backfill.py b/src/fxhnt/application/bybit_paper_backfill.py index 382fc91..38ef093 100644 --- a/src/fxhnt/application/bybit_paper_backfill.py +++ b/src/fxhnt/application/bybit_paper_backfill.py @@ -45,15 +45,19 @@ def _extract_per_sleeve( store: Any, *, universe: set[str] | None, unlock_events: list[dict[str, Any]] | None, sleeves: tuple[str, ...], + positioning_coin_gross_cap: float | None = None, ) -> dict[str, dict[int, dict[str, tuple[float, float]]]]: """`{sleeve: {epoch_day: {symbol: (weight, TRUE return)}}}` — one reconciled engine run per sleeve (`sleeve_weights_and_contributions`, Σ w·r == the edge's GROSS return). The single source the combined WEIGHTS and the per-symbol RETURNS are both derived from, so the live book and the reconstruction - consume identical pairs. A sleeve with no weights is absent. READ-ONLY; memory-bounded by `universe`.""" + consume identical pairs. A sleeve with no weights is absent. `positioning_coin_gross_cap` (positioning + sleeve ONLY) forwards into `sleeve_weights_and_contributions` — default None preserves the byte-identical + uncapped behaviour. READ-ONLY; memory-bounded by `universe`.""" out: dict[str, dict[int, dict[str, tuple[float, float]]]] = {} for sleeve in sleeves: per_day = sleeve_weights_and_contributions(store, sleeve, universe=universe, - unlock_events=unlock_events) + unlock_events=unlock_events, + positioning_coin_gross_cap=positioning_coin_gross_cap) if per_day: out[sleeve] = per_day return out @@ -154,6 +158,7 @@ def backfill_bybit_paper_book( sleeves: tuple[str, ...] = _DEFAULT_BYBIT_SLEEVES, since_day: int | None = None, real_spread_by_coin: dict[str, float] | None = None, + positioning_coin_gross_cap: float | None = None, ) -> int: """Walk every historical Bybit day on which the 4-edge weights exist and persist the live eq-wt book for each day. Returns the number of dates with a persisted nav row. @@ -184,12 +189,17 @@ def backfill_bybit_paper_book( — understating the book ~2.6x (the ~+294% bug vs the honest reconstruction). It now books the TRUE edge returns and reconciles to the reconstruction (guarded by test_bybit_book_reconciliation). + `positioning_coin_gross_cap` (positioning sleeve ONLY) forwards into `_extract_per_sleeve` / + `sleeve_weights_and_contributions` so the backfilled positions honor the same per-coin cap the reconciled + return uses — default None preserves the byte-identical uncapped behaviour. + READ-ONLY against the warehouse `store`; only the paper store is written. The shared `_BYBIT_COST_RATE` keeps the backfill and the live forward track on the SAME cost basis → one continuous book.""" n = len(sleeves) if n == 0: return 0 - per_sleeve = _extract_per_sleeve(store, universe=universe, unlock_events=unlock_events, sleeves=sleeves) + per_sleeve = _extract_per_sleeve(store, universe=universe, unlock_events=unlock_events, sleeves=sleeves, + positioning_coin_gross_cap=positioning_coin_gross_cap) weights_by_day = _combined_weights(per_sleeve, n) if not weights_by_day: return 0 diff --git a/src/fxhnt/cli.py b/src/fxhnt/cli.py index b6f9ba7..87300b4 100644 --- a/src/fxhnt/cli.py +++ b/src/fxhnt/cli.py @@ -3342,6 +3342,7 @@ def bybit_paper_backfill_cmd( from fxhnt.adapters.orchestration.assets import _data_dir from fxhnt.adapters.persistence.paper_repo import PaperRepo from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore + from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP from fxhnt.application.bybit_liquidity import liquid_universe from fxhnt.application.bybit_paper_backfill import backfill_bybit_paper_book from fxhnt.application.bybit_spread_refresh import read_real_spread_by_coin @@ -3376,7 +3377,8 @@ def bybit_paper_backfill_cmd( n = backfill_bybit_paper_book( repo, store, capital=cap, at=dt.datetime.now(dt.UTC), universe=universe, unlock_events=unlock_events, since_day=since_day, - real_spread_by_coin=real_spread) + real_spread_by_coin=real_spread, + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) # PROACTIVE: re-materialize the headline nav-summary for the freshly-backfilled nav so / + /paper are # immediately fast (read 1 row, not the full nav). Guarded — the read-path self-heal is the safety net, # so a refresh failure must never fail the backfill. diff --git a/tests/unit/test_positioning_cap_coverage_audit.py b/tests/unit/test_positioning_cap_coverage_audit.py new file mode 100644 index 0000000..246131c --- /dev/null +++ b/tests/unit/test_positioning_cap_coverage_audit.py @@ -0,0 +1,40 @@ +"""Cap-coverage audit: every LIVE-book / position-sizing / allocation callsite that feeds the positioning +sleeve MUST pass positioning_coin_gross_cap (opt-in, fail-safe). Research/eval callsites are intentionally +uncapped and are excluded. If you add a new LIVE path, add its callsite here AND thread the cap.""" +import re +from pathlib import Path + +_ROOT = Path(__file__).resolve().parents[2] / "src" / "fxhnt" + +# (file, function-that-calls-a-seam) that are LIVE and MUST pass the cap. Grow this list when a new live +# path is added — that is the point: it forces the author to thread the cap. +_LIVE_CALLSITES = [ + ("application/bybit_paper_book.py", "combined_symbol_weights"), + ("application/bybit_paper_book.py", "combined_symbol_weights_and_returns"), + ("application/bybit_paper_book.py", "derive_and_persist_bybit_paper_book"), + ("application/bybit_paper_backfill.py", "backfill_bybit_paper_book"), + ("application/allocation_honest_inputs.py", "honest_allocation_inputs"), + ("application/bybit_testnet_run.py", "run_bybit_testnet"), +] + + +def _func_body(text: str, fn: str) -> str: + # crude but effective: from `def fn(` to the next top-level `def ` / `class ` at column 0. (Anchored on + # `\n(def |class )` rather than a bare `^\S` — with re.MULTILINE, `^` also matches position 0 of the + # sliced remainder, which would truncate the body to the signature line alone.) + m = re.search(rf"^def {re.escape(fn)}\(", text, re.M) + assert m, f"{fn} not found" + start = m.start() + nxt = re.search(r"\n(?:def |class )\S", text[m.end():]) + end = m.end() + nxt.start() + 1 if nxt else len(text) + return text[start:end] + + +def test_every_live_positioning_callsite_threads_the_cap(): + missing = [] + for rel, fn in _LIVE_CALLSITES: + body = _func_body((_ROOT / rel).read_text(), fn) + if "positioning_coin_gross_cap" not in body: + missing.append(f"{rel}::{fn}") + assert not missing, ("LIVE positioning path(s) missing positioning_coin_gross_cap (opt-in, fail-safe): " + + ", ".join(missing)) From 9080dffed5dcfaff5acfc98b68bf60776c1bfaa0 Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 20:32:31 +0000 Subject: [PATCH 15/15] test(positioning): harden cap-coverage audit to require FORWARDING, not mere presence Final whole-branch review Minor: the audit checked the token appeared anywhere in the function body, so a partial regression (param kept in signature but dropped from the inner seam call) passed silently. Now require as a forwarded kwarg. Mutation-verified: dropping the forwarding from run_bybit_testnet's call (while keeping the import) now FAILS the audit, where the substring check passed. Co-Authored-By: Claude Opus 4.8 --- tests/unit/test_positioning_cap_coverage_audit.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/tests/unit/test_positioning_cap_coverage_audit.py b/tests/unit/test_positioning_cap_coverage_audit.py index 246131c..21a9127 100644 --- a/tests/unit/test_positioning_cap_coverage_audit.py +++ b/tests/unit/test_positioning_cap_coverage_audit.py @@ -31,10 +31,15 @@ def _func_body(text: str, fn: str) -> str: def test_every_live_positioning_callsite_threads_the_cap(): + # Require the param to appear as a FORWARDED keyword arg (`positioning_coin_gross_cap=` or the seam's + # `coin_gross_cap=` in bybit_testnet_run's direct-seam calls), not merely SOMEWHERE in the body. A bare + # substring check would pass a partial regression where the param survives in the signature/docstring but + # is dropped from the inner seam call — exactly the silent break this guard exists to catch. + _forwarded = re.compile(r"(?:positioning_)?coin_gross_cap\s*=") missing = [] for rel, fn in _LIVE_CALLSITES: body = _func_body((_ROOT / rel).read_text(), fn) - if "positioning_coin_gross_cap" not in body: + if not _forwarded.search(body): missing.append(f"{rel}::{fn}") - assert not missing, ("LIVE positioning path(s) missing positioning_coin_gross_cap (opt-in, fail-safe): " - + ", ".join(missing)) + assert not missing, ("LIVE positioning path(s) not FORWARDING positioning_coin_gross_cap (opt-in, " + "fail-safe): " + ", ".join(missing))