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

Merge: positioning per-coin gross cap (0.07) — cap every live money-path

Browse Source 
Forensics found the positioning sleeve's 2026-07 -15% drawdown was ~entirely one
coin (LABUSDT, 10.3% long leg, -59%/-79.5% over 2 days). A 1295-day sweep validated
an ex-ante per-coin gross cap of 0.07: Sharpe 2.60->2.69, maxDD -32.3%->-29.9%,
per-year Sharpe non-decreasing; cap=0.04 falsified (2.43). Single-pass clip+renormalize
on the forming weights (causal), validated ~= iterative.

Threaded through EVERY live sink via opt-in (default None = uncapped = fail-safe):
modeled return, reconciliation ref, forward NAV, paper-book positions, testnet exec
sizing, live backfill, capital allocation. Research/eval paths left uncapped by design
(honest falsification). A source-level coverage-audit test (mutation-verified) guards
against a future uncapped live path.

positioning / bybit_4edge / bybit_4edge_levered re-inception v3, inception_t0 backdated
to 2026-07-06 so the forward gate recomputes the window (not reset).

10 tasks, whole-branch review: MERGE. 1126 tests green (the 1 skip-worthy failure is a
pre-existing torch/poc-extra env artifact, identical on master).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Jeroen Grusewski
2026-07-20 20:44:38 +00:00
parent dcf333d97e 9080dffed5
commit 90fcbfde5c
21 changed files with 478 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
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,22 +160,25 @@ 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,
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).
@@ -179,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.
@@ -865,7 +878,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 +886,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 +927,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 +935,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 +969,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 +977,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}, "
@@ -1077,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
@@ -1093,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

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)

40
src/fxhnt/application/allocation_honest_inputs.py
View File

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

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

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

12
src/fxhnt/application/backtest_refs_report.py
View File

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

21
src/fxhnt/application/bybit_book_eval.py
View File

@@ -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,20 +212,20 @@ 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)
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)")

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

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

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

@@ -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):
@@ -135,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
@@ -147,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
@@ -164,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
@@ -208,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
@@ -216,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)

16
src/fxhnt/application/bybit_paper_backfill.py
View File

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

37
src/fxhnt/application/bybit_paper_book.py
View File

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

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

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

@@ -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
@@ -140,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:
@@ -174,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)
@@ -247,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
@@ -262,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):
@@ -288,6 +305,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
@@ -295,9 +313,12 @@ 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."""
return _book_breakdown(store, universe=universe, cost_bps=cost_bps, direction=direction,
capacity_capital=capacity_capital, tail_cap_k=tail_cap_k)["net"]
`_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`)."""
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

10
src/fxhnt/application/bybit_testnet_run.py
View File

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

8
src/fxhnt/cli.py
View File

@@ -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()
@@ -3340,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
@@ -3374,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.

12
src/fxhnt/registry.py
View File

@@ -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: 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
@@ -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: coin_gross_cap=0.07 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: coin_gross_cap=0.07 backfill
"backtest": {"kind": "report"}},
},
# NOTE: `xvenue_carry` (cross-venue Bybit↔Deribit static carry) was RETIRED from forward-tracking

60
tests/integration/test_bybit_paper_reconciliation.py
View File

@@ -122,3 +122,63 @@ 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"
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}"

111
tests/integration/test_bybit_positioning_eval.py
View File

@@ -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:
# Realistic ~60-coin panel with one dominant over-long coin (LABUSDT-class). Uncapped, the dominant coin's
# |weight| exceeds the 0.07 cap; capped, it is pulled down close to the cap (single-pass clip+renormalize
# binds strongly when many coins carry weight). Verifies the cap substantially reduces concentration,
# preserves unit gross, and keeps the sign structure.
ratio = {0: {f"C{i}USDT": 0.30 + 0.0067 * i for i in range(60)}} # 60 coins spread from 0.30 to ~0.70
ratio[0]["DOMUSDT"] = 1.10 # the extreme over-long coin (fade -> short)
uncapped = positioning_weights(ratio)[0]
capped = positioning_weights(ratio, coin_gross_cap=0.07)[0]
assert abs(uncapped["DOMUSDT"]) > 0.07 # dominant coin exceeds the cap uncapped
assert abs(capped["DOMUSDT"]) < abs(uncapped["DOMUSDT"]) # cap pulls it down
assert abs(capped["DOMUSDT"]) < 0.085 # close to the 0.07 cap (single-pass slack)
assert abs(sum(abs(w) for w in capped.values()) - 1.0) < 1e-9 # still unit gross
assert capped["DOMUSDT"] < 0.0 # over-long crowd -> short (sign preserved)
def test_coin_gross_cap_none_is_byte_identical() -> None:
ratio = {0: {"AAAUSDT": 0.9, "BBBUSDT": 0.5, "CCCUSDT": 0.1}}
assert positioning_weights(ratio, coin_gross_cap=None) == positioning_weights(ratio)
# --- 2. positioning_metrics_from_store ---------------------------------------------------------
def _seed_contrarian(store: TimescaleFeatureStore, *, days: int = 80) -> None:
@@ -352,3 +373,93 @@ 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 constant and parameter threading through book_breakdown + returns_from_store ------
def test_positioning_coin_gross_cap_constant_is_007() -> None:
from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP
assert POSITIONING_COIN_GROSS_CAP == 0.07
def test_coin_gross_cap_reduces_single_coin_loss_in_book() -> None:
"""Shock store: one coin (retail over-short at extreme long_ratio ~0.05) craters ~50% the next day.
Uncapped, that coin's loss dominates the book loss. Capped at 0.07, its weight is bounded,
reducing (making less bad) the worst-day loss."""
from fxhnt.application.bybit_positioning_eval import positioning_returns_from_store
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
# Build shock store: 4 coins, 1 at extreme long_ratio (retail over-short), crash next day
# Days 0-2: normal, Day 2->3: crash day for LABUSDT
px = {"LABUSDT": 100.0, "AAAUSDT": 100.0, "BBBUSDT": 100.0, "CCCUSDT": 100.0}
for d in range(4):
# LABUSDT is over-short (low long_ratio ~0.05), others are near median ~0.5
store.write_features("LABUSDT", [(d * _DAY, {"long_ratio": 0.05, "close": px["LABUSDT"]})])
store.write_features("AAAUSDT", [(d * _DAY, {"long_ratio": 0.50, "close": px["AAAUSDT"]})])
store.write_features("BBBUSDT", [(d * _DAY, {"long_ratio": 0.48, "close": px["BBBUSDT"]})])
store.write_features("CCCUSDT", [(d * _DAY, {"long_ratio": 0.52, "close": px["CCCUSDT"]})])
# Normal prices on days 0, 1, 2
if d < 2:
px["LABUSDT"] *= 1.00
px["AAAUSDT"] *= 1.00
px["BBBUSDT"] *= 1.00
px["CCCUSDT"] *= 1.00
# LABUSDT crashes ~50% on the transition day 2->3 (its long_ratio on day 2 forms the weight)
elif d == 2:
px["LABUSDT"] *= 0.50 # crash next day
px["AAAUSDT"] *= 1.00
px["BBBUSDT"] *= 1.00
px["CCCUSDT"] *= 1.00
# (day 3: no more transitions)
uncapped = positioning_returns_from_store(store, cost_bps=0.0)
capped = positioning_returns_from_store(store, cost_bps=0.0, coin_gross_cap=0.07)
store.close()
# The store MUST produce a book (empty dicts would mean the fixture is broken — assert loudly, never
# skip the real assertion silently).
assert uncapped and capped, "shock store produced no positioning returns — fixture broken"
# Worst day = the crash day (day 2: weight formed, day 2->3 loss realized). The cap bounds the per-coin
# weight, making the worst-day loss less bad (less negative).
shock_day = min(uncapped, key=lambda d: uncapped[d])
assert capped[shock_day] > uncapped[shock_day], \
f"cap should improve worst day: uncapped={uncapped[shock_day]}, capped={capped[shock_day]}"
def test_sleeve_returns_positioning_forwards_coin_gross_cap() -> None:
"""sleeve_returns_from_store forwards coin_gross_cap to the positioning edge only."""
from fxhnt.application.bybit_book_eval import sleeve_returns_from_store
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
# Build shock store: 4 coins, 1 at extreme long_ratio (retail over-short), crash next day
px = {"LABUSDT": 100.0, "AAAUSDT": 100.0, "BBBUSDT": 100.0, "CCCUSDT": 100.0}
for d in range(4):
store.write_features("LABUSDT", [(d * _DAY, {"long_ratio": 0.05, "close": px["LABUSDT"]})])
store.write_features("AAAUSDT", [(d * _DAY, {"long_ratio": 0.50, "close": px["AAAUSDT"]})])
store.write_features("BBBUSDT", [(d * _DAY, {"long_ratio": 0.48, "close": px["BBBUSDT"]})])
store.write_features("CCCUSDT", [(d * _DAY, {"long_ratio": 0.52, "close": px["CCCUSDT"]})])
if d < 2:
px["LABUSDT"] *= 1.00
px["AAAUSDT"] *= 1.00
px["BBBUSDT"] *= 1.00
px["CCCUSDT"] *= 1.00
elif d == 2:
px["LABUSDT"] *= 0.50
px["AAAUSDT"] *= 1.00
px["BBBUSDT"] *= 1.00
px["CCCUSDT"] *= 1.00
uncapped = sleeve_returns_from_store(store, "positioning", cost_bps=0.0)
capped = sleeve_returns_from_store(store, "positioning", cost_bps=0.0, coin_gross_cap=0.07)
store.close()
# The store MUST produce a book (empty dicts would mean the seam or fixture is broken — assert loudly,
# never skip the real assertion silently).
assert uncapped and capped, "shock store produced no positioning returns via the sleeve seam — broken"
# cap forwarded through the sleeve seam -> worst day STRICTLY improved (the deterministic shock always
# yields a strict improvement: −0.25 -> −0.125). Strict `>` so a silent no-op (cap not forwarded) fails.
shock_day = min(uncapped, key=lambda d: uncapped[d])
assert capped[shock_day] > uncapped[shock_day], \
f"cap should improve worst day: uncapped={uncapped[shock_day]}, capped={capped[shock_day]}"

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

4
tests/unit/test_forward_definition.py
View File

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

45
tests/unit/test_positioning_cap_coverage_audit.py Normal file
View File

@@ -0,0 +1,45 @@
"""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():
# 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 not _forwarded.search(body):
missing.append(f"{rel}::{fn}")
assert not missing, ("LIVE positioning path(s) not FORWARDING positioning_coin_gross_cap (opt-in, "
"fail-safe): " + ", ".join(missing))

11
tests/unit/test_positioning_cap_reinception.py Normal file
View File

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