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

fix(positioning): cap the last 2 live paths (paper-backfill + allocation) + coverage-audit regression test

Browse Source 
An exhaustive sweep of every caller in the positioning-cap dependency chain found two more live paths
still uncapped: the Bybit paper-book BACKFILL (bybit_paper_backfill.py, sizes/persists live paper
positions across history) and the honest-reference allocation inputs (allocation_honest_inputs.py,
feeds record_allocation, the LIVE strategy allocation engine's ex-ante capital sizing).

Threads positioning_coin_gross_cap (opt-in, default None) through both, matching the existing pattern:
- bybit_paper_backfill.py: _extract_per_sleeve + backfill_bybit_paper_book now accept and forward the
  cap; cli.py's bybit-paper-backfill command passes POSITIONING_COIN_GROSS_CAP.
- allocation_honest_inputs.py: _sleeve_inputs + _book_curve + _deploy_curve + honest_allocation_inputs
  now accept and forward the cap; allocation_ingest.py's record_allocation (called from the
  cockpit_forward Dagster asset) passes POSITIONING_COIN_GROSS_CAP into honest_allocation_inputs.

Adds tests/unit/test_positioning_cap_coverage_audit.py — a source-level regression guard asserting every
LIVE-book/position/allocation callsite passes positioning_coin_gross_cap. Fixed a boundary bug in the
brief's own _func_body helper regex (bare ^\S with re.MULTILINE always matches position 0 of the sliced
remainder, truncating every function body to its signature line) so the audit actually inspects each
function's body instead of false-failing on already-capped callsites.

Research/eval callers (verify_positioning_edge, walk_forward_positioning, look_ahead_audit,
_variant_weights, _drop_top_n, _liquidity_sweep, positioning_metalabel, bybit_overlay_ab, vrp_eval,
onchain_fundamental_eval, spread_overlay, honest_report) are untouched — they stay uncapped by design.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Jeroen Grusewski
2026-07-20 20:17:20 +00:00
parent f325a37070
commit 3a26b87f15
5 changed files with 87 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

4
src/fxhnt/cli.py
View File

@@ -3342,6 +3342,7 @@ def bybit_paper_backfill_cmd(
from fxhnt.adapters.orchestration.assets import _data_dir
from fxhnt.adapters.persistence.paper_repo import PaperRepo
from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP
from fxhnt.application.bybit_liquidity import liquid_universe
from fxhnt.application.bybit_paper_backfill import backfill_bybit_paper_book
from fxhnt.application.bybit_spread_refresh import read_real_spread_by_coin
@@ -3376,7 +3377,8 @@ def bybit_paper_backfill_cmd(
n = backfill_bybit_paper_book(
repo, store, capital=cap, at=dt.datetime.now(dt.UTC),
universe=universe, unlock_events=unlock_events, since_day=since_day,
real_spread_by_coin=real_spread)
real_spread_by_coin=real_spread,
positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP)
# PROACTIVE: re-materialize the headline nav-summary for the freshly-backfilled nav so / + /paper are
# immediately fast (read 1 row, not the full nav). Guarded — the read-path self-heal is the safety net,
# so a refresh failure must never fail the backfill.

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

@@ -0,0 +1,40 @@
"""Cap-coverage audit: every LIVE-book / position-sizing / allocation callsite that feeds the positioning
sleeve MUST pass positioning_coin_gross_cap (opt-in, fail-safe). Research/eval callsites are intentionally
uncapped and are excluded. If you add a new LIVE path, add its callsite here AND thread the cap."""
import re
from pathlib import Path
_ROOT = Path(__file__).resolve().parents[2] / "src" / "fxhnt"
# (file, function-that-calls-a-seam) that are LIVE and MUST pass the cap. Grow this list when a new live
# path is added — that is the point: it forces the author to thread the cap.
_LIVE_CALLSITES = [
("application/bybit_paper_book.py", "combined_symbol_weights"),
("application/bybit_paper_book.py", "combined_symbol_weights_and_returns"),
("application/bybit_paper_book.py", "derive_and_persist_bybit_paper_book"),
("application/bybit_paper_backfill.py", "backfill_bybit_paper_book"),
("application/allocation_honest_inputs.py", "honest_allocation_inputs"),
("application/bybit_testnet_run.py", "run_bybit_testnet"),
]
def _func_body(text: str, fn: str) -> str:
# crude but effective: from `def fn(` to the next top-level `def ` / `class ` at column 0. (Anchored on
# `\n(def |class )` rather than a bare `^\S` — with re.MULTILINE, `^` also matches position 0 of the
# sliced remainder, which would truncate the body to the signature line alone.)
m = re.search(rf"^def {re.escape(fn)}\(", text, re.M)
assert m, f"{fn} not found"
start = m.start()
nxt = re.search(r"\n(?:def |class )\S", text[m.end():])
end = m.end() + nxt.start() + 1 if nxt else len(text)
return text[start:end]
def test_every_live_positioning_callsite_threads_the_cap():
missing = []
for rel, fn in _LIVE_CALLSITES:
body = _func_body((_ROOT / rel).read_text(), fn)
if "positioning_coin_gross_cap" not in body:
missing.append(f"{rel}::{fn}")
assert not missing, ("LIVE positioning path(s) missing positioning_coin_gross_cap (opt-in, fail-safe): "
+ ", ".join(missing))