gitadmin/fxhnt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(positioning): thread coin_gross_cap into the LIVE forward-nav path + attribution (final-review Critical)

Browse Source 
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).
This commit is contained in:
Jeroen Grusewski
2026-07-20 19:24:44 +00:00
parent f2730d195d
commit 5d4ddcfb39
7 changed files with 120 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
src/fxhnt/adapters/orchestration/assets.py
View File

@@ -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}, "

22
src/fxhnt/adapters/orchestration/migration_builders.py
View File

@@ -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)

5
src/fxhnt/application/bybit_forward_track.py
View File

@@ -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

9
src/fxhnt/application/bybit_paper_weights.py
View File

@@ -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})")

4
src/fxhnt/application/bybit_positioning_eval.py
View File

@@ -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):

37
tests/integration/test_bybit_paper_reconciliation.py
View File

@@ -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"

27
tests/integration/test_positioning_forward_track.py
View File

@@ -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]}"