From 3a26b87f15e5173b860bcfd0ff7732ca79420204 Mon Sep 17 00:00:00 2001 From: Jeroen Grusewski Date: Mon, 20 Jul 2026 20:17:20 +0000 Subject: [PATCH] fix(positioning): cap the last 2 live paths (paper-backfill + allocation) + coverage-audit regression test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An exhaustive sweep of every caller in the positioning-cap dependency chain found two more live paths still uncapped: the Bybit paper-book BACKFILL (bybit_paper_backfill.py, sizes/persists live paper positions across history) and the honest-reference allocation inputs (allocation_honest_inputs.py, feeds record_allocation, the LIVE strategy allocation engine's ex-ante capital sizing). Threads positioning_coin_gross_cap (opt-in, default None) through both, matching the existing pattern: - bybit_paper_backfill.py: _extract_per_sleeve + backfill_bybit_paper_book now accept and forward the cap; cli.py's bybit-paper-backfill command passes POSITIONING_COIN_GROSS_CAP. - allocation_honest_inputs.py: _sleeve_inputs + _book_curve + _deploy_curve + honest_allocation_inputs now accept and forward the cap; allocation_ingest.py's record_allocation (called from the cockpit_forward Dagster asset) passes POSITIONING_COIN_GROSS_CAP into honest_allocation_inputs. Adds tests/unit/test_positioning_cap_coverage_audit.py — a source-level regression guard asserting every LIVE-book/position/allocation callsite passes positioning_coin_gross_cap. Fixed a boundary bug in the brief's own _func_body helper regex (bare ^\S with re.MULTILINE always matches position 0 of the sliced remainder, truncating every function body to its signature line) so the audit actually inspects each function's body instead of false-failing on already-capped callsites. Research/eval callers (verify_positioning_edge, walk_forward_positioning, look_ahead_audit, _variant_weights, _drop_top_n, _liquidity_sweep, positioning_metalabel, bybit_overlay_ab, vrp_eval, onchain_fundamental_eval, spread_overlay, honest_report) are untouched — they stay uncapped by design. Co-Authored-By: Claude Opus 4.8 --- .../application/allocation_honest_inputs.py | 40 +++++++++++++------ src/fxhnt/application/allocation_ingest.py | 4 +- src/fxhnt/application/bybit_paper_backfill.py | 16 ++++++-- src/fxhnt/cli.py | 4 +- .../test_positioning_cap_coverage_audit.py | 40 +++++++++++++++++++ 5 files changed, 87 insertions(+), 17 deletions(-) create mode 100644 tests/unit/test_positioning_cap_coverage_audit.py diff --git a/src/fxhnt/application/allocation_honest_inputs.py b/src/fxhnt/application/allocation_honest_inputs.py index 8364b7c..e8e97ec 100644 --- a/src/fxhnt/application/allocation_honest_inputs.py +++ b/src/fxhnt/application/allocation_honest_inputs.py @@ -68,11 +68,15 @@ def _iso(epoch_day: int) -> str: def _sleeve_inputs(store: Any, sleeve: str, - unlock_events: list[dict[str, Any]] | None) -> tuple[dict, dict] | None: + unlock_events: list[dict[str, Any]] | None, + positioning_coin_gross_cap: float | None = None) -> tuple[dict, dict] | None: """AUM-INDEPENDENT heavy inputs for a sleeve: (weights_by_day, turnover_panel). Load ONCE, reprice across AUMs via honest_sleeve_returns/capacity_curve — the turnover panel read is the memory cost, so it must NOT - be reloaded per AUM. None when the sleeve is uncomputable (e.g. unlock with no events). READ-ONLY.""" - wbd = sleeve_weights_and_contributions(store, sleeve, unlock_events=unlock_events) + be reloaded per AUM. None when the sleeve is uncomputable (e.g. unlock with no events). + `positioning_coin_gross_cap` (positioning sleeve ONLY) forwards into `sleeve_weights_and_contributions` — + default None preserves the byte-identical uncapped behaviour. READ-ONLY.""" + wbd = sleeve_weights_and_contributions(store, sleeve, unlock_events=unlock_events, + positioning_coin_gross_cap=positioning_coin_gross_cap) if not wbd: return None symbols = sorted({s for row in wbd.values() for s in row}) @@ -81,11 +85,14 @@ def _sleeve_inputs(store: Any, sleeve: str, def _book_curve(store: Any, spread_bps: dict[str, float], unlock_events: list[dict[str, Any]] | None, - aums, *, participation_cap: float, impact_k: float) -> dict[float, dict[int, float]]: + aums, *, participation_cap: float, impact_k: float, + positioning_coin_gross_cap: float | None = None) -> dict[float, dict[int, float]]: """{aum: risk-parity BOOK series} across `aums`, loading each `_BOOK_SLEEVES` sleeve's inputs ONCE then - repricing per AUM. Empty sleeves (uncomputable) are dropped; an AUM with no computable sleeve -> {}.""" + repricing per AUM. Empty sleeves (uncomputable) are dropped; an AUM with no computable sleeve -> {}. + `positioning_coin_gross_cap` forwards into `_sleeve_inputs` (positioning sleeve ONLY).""" loaded = {sleeve: inp for sleeve in _BOOK_SLEEVES - if (inp := _sleeve_inputs(store, sleeve, unlock_events)) is not None} + if (inp := _sleeve_inputs(store, sleeve, unlock_events, + positioning_coin_gross_cap=positioning_coin_gross_cap)) is not None} out: dict[float, dict[int, float]] = {} for aum in aums: per: dict[str, dict[int, float]] = {} @@ -100,14 +107,18 @@ def _book_curve(store: Any, spread_bps: dict[str, float], unlock_events: list[di def _deploy_curve(store: Any, sid: str, spread_bps: dict[str, float], unlock_events: list[dict[str, Any]] | None, aums, *, - participation_cap: float, impact_k: float) -> dict[float, dict[int, float]]: + participation_cap: float, impact_k: float, + positioning_coin_gross_cap: float | None = None) -> dict[float, dict[int, float]]: """{aum: honest series} for a crypto deploy `sid` across `aums`, loading AUM-independent inputs ONCE. - `bybit_4edge` -> the risk-parity book; a sleeve sid -> its own series (reuses `capacity_curve`); other -> {}.""" + `bybit_4edge` -> the risk-parity book; a sleeve sid -> its own series (reuses `capacity_curve`); other -> {}. + `positioning_coin_gross_cap` forwards through (positioning sleeve ONLY).""" if sid == "bybit_4edge": return _book_curve(store, spread_bps, unlock_events, aums, - participation_cap=participation_cap, impact_k=impact_k) + participation_cap=participation_cap, impact_k=impact_k, + positioning_coin_gross_cap=positioning_coin_gross_cap) if sid in _CRYPTO_DEPLOY_SIDS: - inputs = _sleeve_inputs(store, sid, unlock_events) + inputs = _sleeve_inputs(store, sid, unlock_events, + positioning_coin_gross_cap=positioning_coin_gross_cap) if inputs is None: return {aum: {} for aum in aums} wbd, turnover = inputs @@ -120,6 +131,7 @@ def honest_allocation_inputs( store: Any, spread_bps: dict[str, float], unlock_events: list[dict[str, Any]] | None, *, deploy_sids, target_aum: float, aums, participation_cap: float = 0.03, impact_k: float = 1.0, min_sharpe: float = 1.0, equity_default_ceiling: float = 10_000_000.0, + positioning_coin_gross_cap: float | None = None, ) -> tuple[dict[str, dict[str, float]], dict[str, float | None]]: """`(returns_by_strategy, capacity_ceiling_by_strategy)` for the deploy set — the allocation engine's honest inputs. @@ -127,7 +139,10 @@ def honest_allocation_inputs( For each crypto deploy sid: its honest series at `target_aum` (returned STRING-keyed, see the module docstring seam) plus a capacity ceiling = `capacity_ceiling_from_curve({aum: sharpe_of(series(aum)) for aum in aums}, min_sharpe)`. Non-crypto sids get `equity_default_ceiling` and NO series (the caller - supplies their reference). PURE + read-only.""" + supplies their reference). `positioning_coin_gross_cap` (positioning sleeve ONLY) forwards into + `_deploy_curve` / `_sleeve_inputs` so the honest-reference basis the LIVE allocation engine + (`record_allocation`) sizes against honors the same per-coin cap the live capped return uses — default + None preserves the byte-identical uncapped behaviour. PURE + read-only.""" returns_by_strategy: dict[str, dict[str, float]] = {} ceilings: dict[str, float | None] = {} for sid in deploy_sids: @@ -137,7 +152,8 @@ def honest_allocation_inputs( # AUM-independent panels loaded ONCE inside _deploy_curve; base + curve are repriced from them. all_aums = sorted({float(target_aum), *(float(a) for a in aums)}) curve_series = _deploy_curve(store, sid, spread_bps, unlock_events, all_aums, - participation_cap=participation_cap, impact_k=impact_k) + participation_cap=participation_cap, impact_k=impact_k, + positioning_coin_gross_cap=positioning_coin_gross_cap) base = curve_series.get(float(target_aum), {}) if base: # int epoch-day -> ISO date string, the SAME key the nav fallback uses, consistently across every diff --git a/src/fxhnt/application/allocation_ingest.py b/src/fxhnt/application/allocation_ingest.py index d71fb4c..e1a2ac0 100644 --- a/src/fxhnt/application/allocation_ingest.py +++ b/src/fxhnt/application/allocation_ingest.py @@ -22,6 +22,7 @@ from fxhnt.application.allocation_policy import ( allocation_policy_hash, resolve_allocation_policy, ) +from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP from fxhnt.application.forward_ingest import CockpitBacktestRefProvider from fxhnt.application.funding_status import FundingStatus, compute_funding from fxhnt.application.gate_policy import POLICY, POLICY_VERSION, policy_hash @@ -89,7 +90,8 @@ def record_allocation(dsn: str, equity: float, *, store=None, spread_bps=None, deploy_sids=funded_sids, target_aum=equity, aums=[35_000, 100_000, 350_000, 1_000_000, 10_000_000], min_sharpe=getattr(pol, "capacity_min_sharpe", 1.0), - equity_default_ceiling=getattr(pol, "equity_default_ceiling", 10_000_000.0)) + equity_default_ceiling=getattr(pol, "equity_default_ceiling", 10_000_000.0), + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) # Equity leg (sub-project: equity leg deployable, Task 3): the ibkr-equity deploy sids (today only # `multistrat`) get their OWN cost-netted sizing series + real-ADV capacity ceiling here, merged on diff --git a/src/fxhnt/application/bybit_paper_backfill.py b/src/fxhnt/application/bybit_paper_backfill.py index 382fc91..38ef093 100644 --- a/src/fxhnt/application/bybit_paper_backfill.py +++ b/src/fxhnt/application/bybit_paper_backfill.py @@ -45,15 +45,19 @@ def _extract_per_sleeve( store: Any, *, universe: set[str] | None, unlock_events: list[dict[str, Any]] | None, sleeves: tuple[str, ...], + positioning_coin_gross_cap: float | None = None, ) -> dict[str, dict[int, dict[str, tuple[float, float]]]]: """`{sleeve: {epoch_day: {symbol: (weight, TRUE return)}}}` — one reconciled engine run per sleeve (`sleeve_weights_and_contributions`, Σ w·r == the edge's GROSS return). The single source the combined WEIGHTS and the per-symbol RETURNS are both derived from, so the live book and the reconstruction - consume identical pairs. A sleeve with no weights is absent. READ-ONLY; memory-bounded by `universe`.""" + consume identical pairs. A sleeve with no weights is absent. `positioning_coin_gross_cap` (positioning + sleeve ONLY) forwards into `sleeve_weights_and_contributions` — default None preserves the byte-identical + uncapped behaviour. READ-ONLY; memory-bounded by `universe`.""" out: dict[str, dict[int, dict[str, tuple[float, float]]]] = {} for sleeve in sleeves: per_day = sleeve_weights_and_contributions(store, sleeve, universe=universe, - unlock_events=unlock_events) + unlock_events=unlock_events, + positioning_coin_gross_cap=positioning_coin_gross_cap) if per_day: out[sleeve] = per_day return out @@ -154,6 +158,7 @@ def backfill_bybit_paper_book( sleeves: tuple[str, ...] = _DEFAULT_BYBIT_SLEEVES, since_day: int | None = None, real_spread_by_coin: dict[str, float] | None = None, + positioning_coin_gross_cap: float | None = None, ) -> int: """Walk every historical Bybit day on which the 4-edge weights exist and persist the live eq-wt book for each day. Returns the number of dates with a persisted nav row. @@ -184,12 +189,17 @@ def backfill_bybit_paper_book( — understating the book ~2.6x (the ~+294% bug vs the honest reconstruction). It now books the TRUE edge returns and reconciles to the reconstruction (guarded by test_bybit_book_reconciliation). + `positioning_coin_gross_cap` (positioning sleeve ONLY) forwards into `_extract_per_sleeve` / + `sleeve_weights_and_contributions` so the backfilled positions honor the same per-coin cap the reconciled + return uses — default None preserves the byte-identical uncapped behaviour. + READ-ONLY against the warehouse `store`; only the paper store is written. The shared `_BYBIT_COST_RATE` keeps the backfill and the live forward track on the SAME cost basis → one continuous book.""" n = len(sleeves) if n == 0: return 0 - per_sleeve = _extract_per_sleeve(store, universe=universe, unlock_events=unlock_events, sleeves=sleeves) + per_sleeve = _extract_per_sleeve(store, universe=universe, unlock_events=unlock_events, sleeves=sleeves, + positioning_coin_gross_cap=positioning_coin_gross_cap) weights_by_day = _combined_weights(per_sleeve, n) if not weights_by_day: return 0 diff --git a/src/fxhnt/cli.py b/src/fxhnt/cli.py index b6f9ba7..87300b4 100644 --- a/src/fxhnt/cli.py +++ b/src/fxhnt/cli.py @@ -3342,6 +3342,7 @@ def bybit_paper_backfill_cmd( from fxhnt.adapters.orchestration.assets import _data_dir from fxhnt.adapters.persistence.paper_repo import PaperRepo from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore + from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP from fxhnt.application.bybit_liquidity import liquid_universe from fxhnt.application.bybit_paper_backfill import backfill_bybit_paper_book from fxhnt.application.bybit_spread_refresh import read_real_spread_by_coin @@ -3376,7 +3377,8 @@ def bybit_paper_backfill_cmd( n = backfill_bybit_paper_book( repo, store, capital=cap, at=dt.datetime.now(dt.UTC), universe=universe, unlock_events=unlock_events, since_day=since_day, - real_spread_by_coin=real_spread) + real_spread_by_coin=real_spread, + positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP) # PROACTIVE: re-materialize the headline nav-summary for the freshly-backfilled nav so / + /paper are # immediately fast (read 1 row, not the full nav). Guarded — the read-path self-heal is the safety net, # so a refresh failure must never fail the backfill. diff --git a/tests/unit/test_positioning_cap_coverage_audit.py b/tests/unit/test_positioning_cap_coverage_audit.py new file mode 100644 index 0000000..246131c --- /dev/null +++ b/tests/unit/test_positioning_cap_coverage_audit.py @@ -0,0 +1,40 @@ +"""Cap-coverage audit: every LIVE-book / position-sizing / allocation callsite that feeds the positioning +sleeve MUST pass positioning_coin_gross_cap (opt-in, fail-safe). Research/eval callsites are intentionally +uncapped and are excluded. If you add a new LIVE path, add its callsite here AND thread the cap.""" +import re +from pathlib import Path + +_ROOT = Path(__file__).resolve().parents[2] / "src" / "fxhnt" + +# (file, function-that-calls-a-seam) that are LIVE and MUST pass the cap. Grow this list when a new live +# path is added — that is the point: it forces the author to thread the cap. +_LIVE_CALLSITES = [ + ("application/bybit_paper_book.py", "combined_symbol_weights"), + ("application/bybit_paper_book.py", "combined_symbol_weights_and_returns"), + ("application/bybit_paper_book.py", "derive_and_persist_bybit_paper_book"), + ("application/bybit_paper_backfill.py", "backfill_bybit_paper_book"), + ("application/allocation_honest_inputs.py", "honest_allocation_inputs"), + ("application/bybit_testnet_run.py", "run_bybit_testnet"), +] + + +def _func_body(text: str, fn: str) -> str: + # crude but effective: from `def fn(` to the next top-level `def ` / `class ` at column 0. (Anchored on + # `\n(def |class )` rather than a bare `^\S` — with re.MULTILINE, `^` also matches position 0 of the + # sliced remainder, which would truncate the body to the signature line alone.) + m = re.search(rf"^def {re.escape(fn)}\(", text, re.M) + assert m, f"{fn} not found" + start = m.start() + nxt = re.search(r"\n(?:def |class )\S", text[m.end():]) + end = m.end() + nxt.start() + 1 if nxt else len(text) + return text[start:end] + + +def test_every_live_positioning_callsite_threads_the_cap(): + missing = [] + for rel, fn in _LIVE_CALLSITES: + body = _func_body((_ROOT / rel).read_text(), fn) + if "positioning_coin_gross_cap" not in body: + missing.append(f"{rel}::{fn}") + assert not missing, ("LIVE positioning path(s) missing positioning_coin_gross_cap (opt-in, fail-safe): " + + ", ".join(missing))