From 03c356d5d5d755e44ced95b3b2cdd6c779a1b58c Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Fri, 26 Jun 2026 14:44:30 +0200 Subject: [PATCH] =?UTF-8?q?feat(research):=20broaden+overlay=20deploy-book?= =?UTF-8?q?=20validation=20=E2=80=94=20walk-forward=20+=20capacity=20head-?= =?UTF-8?q?to-head=20vs=20liquid-61?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Walk-forward stress-test of Step 3's single-split lead: is a BROAD universe + spread-aware overlay a deploy-book UPGRADE over the live liquid-61 ($10M) Bybit book? READ-ONLY SHADOW eval — never mutates the live book. - broaden_overlay_validate.py: expanding-window folds (knobs+breadth selected on TRAIN, evaluated on disjoint TEST → no leakage); per-fold head-to-head live vs broad+overlay NET; bottleneck deployable-capital estimator (capacity ratio invariant to participation_frac); corr + concentration robustness; verdict UPGRADE / CAPACITY-TRADEOFF / OVERFIT / NO. - CLI `fxhnt broaden-overlay-validate` (--min-dollar-vols, --live-bound, --n-folds, --participation-frac): per-fold table + aggregate + capacity verdict. - Reuses spread_overlay (engine + live per-coin weights), bybit_liquidity_sweep (metrics), liquid_universe. 17 TDD tests (no-leakage folds, baseline reconciliation, capacity monotonicity, verdict falsification, WriteTripwire, determinism, CLI). Full suite 1593 passed. Co-Authored-By: Claude Opus 4.8 --- .../application/broaden_overlay_validate.py | 369 ++++++++++++++++++ src/fxhnt/cli.py | 116 ++++++ .../test_broaden_overlay_validate.py | 297 ++++++++++++++ 3 files changed, 782 insertions(+) create mode 100644 src/fxhnt/application/broaden_overlay_validate.py create mode 100644 tests/integration/test_broaden_overlay_validate.py diff --git a/src/fxhnt/application/broaden_overlay_validate.py b/src/fxhnt/application/broaden_overlay_validate.py new file mode 100644 index 0000000..c9b614c --- /dev/null +++ b/src/fxhnt/application/broaden_overlay_validate.py @@ -0,0 +1,369 @@ +"""WALK-FORWARD validation of the DEPLOY-BOOK question Step 3 only hinted at: is a BROAD universe + the +spread-aware execution overlay an UPGRADE over the current liquid-61 ($10M) Bybit book? READ-ONLY, SHADOW +eval (never mutates the live book), network-free, memory-bounded, pure given the store. + +WHY (the lead + the risk) +------------------------- +Step 3 (`spread_overlay`) found, on ONE OOS split with TRAIN-tuned k/λ, that a broad universe + the overlay +BEAT the live book: $10M liquid-61 NET-OOS Sharpe 1.04 / cost-drag 67.8bp vs $1M broad+overlay 1.30 / +28.7bp. ONE split with selected knobs is exactly where overfitting hides. This module stress-tests that lead +with WALK-FORWARD folds so we learn whether the 1.30 is a real, repeatable edge or a one-split fluke — and, +critically, whether the broad book can actually hold deployable capital (the $10M bound exists for a reason). + +THE WALK-FORWARD DESIGN (the rigor Step 3 lacked) +------------------------------------------------- +EXPANDING-WINDOW folds over the full day axis. The axis (sorted union of every candidate universe's days) is +cut into `n_folds + 1` contiguous index blocks; fold i TRAINS on every day up to the end of block i and TESTS +on the disjoint next block. Per fold: + * SELECTION (on TRAIN only): pick BOTH the overlay knobs (k, λ) AND the universe breadth (`min_dollar_vol`) + that maximise the broad book's TRAIN NET Sharpe. No future day touches the choice. + * VERDICT (on the held-out TEST only): evaluate that frozen selection NET on the disjoint test block. +Test blocks are disjoint and every fold's train strictly precedes its test (train_hi < test_lo) → no leakage. +We aggregate the OOS test results across folds (mean NET Sharpe + dispersion + how often broad beats live), +which is what kills the single-split overfit. + +HEAD-TO-HEAD (same test windows) +-------------------------------- +On EVERY fold's TEST window we compare, NET of the MEASURED per-coin cost (Bybit 5.5bp taker + half the +Corwin–Schultz spread): + (A) LIVE book — liquid-61 ($10M `min_dollar_vol`), naive eq-wt per-coin (k=0, λ=0). The current deploy. + (B) BROAD+overlay — the per-fold-train-selected broader universe + spread-aware overlay (k*, λ*). +Both report NET Sharpe / total return / cost-drag / maxDD / turnover, plus Δ(B−A), per fold and aggregated. + +CAPACITY (the $10M bound exists for a reason) +--------------------------------------------- +A higher-Sharpe book that can only hold $200k is NOT an upgrade for a fund that wants to scale. We estimate +each book's DEPLOYABLE CAPITAL from the BOTTLENECK coin: to deploy capital C the book must put |w_i|·C into +coin i, which can absorb at most `participation_frac · ADV_i` (a fraction of the coin's median daily +dollar-volume). So C ≤ participation_frac·ADV_i / |w_i| for every held coin, and the deployable capital is the +MIN over coins of that bound — the thinnest-relative-to-weight name. The overlay DOWNWEIGHTS thin coins (lower +|w_i|), which RAISES their bound and so lifts the book's capacity; capacity therefore reflects the overlay. The +capacity RATIO (B/A) is invariant to `participation_frac` (it cancels), so the verdict's capacity test does not +depend on that assumption. We also report the gross-weighted absorbable notional `Σ participation_frac·ADV_i· +|w_i|` (the brief's sketch) as a supplementary figure. + +VERDICT +------- + * UPGRADE — aggregate walk-forward NET Sharpe lift >= +0.20, broad beats live in a majority (>=60%) + of folds, AND capacity is acceptable (B >= 50% of A's deployable capital). Wire it in. + * CAPACITY-TRADEOFF— same Sharpe win, but B's deployable capital is materially below A's. A choice, not a + free win. + * OVERFIT — a positive but small/inconsistent lift: the 1.30 shrinks toward liquid-61 under + walk-forward (Step 3 was a one-split fluke). Keep liquid-61. + * NO — broad+overlay is no better on average. Keep liquid-61. + +REUSE: `spread_overlay.{per_coin_weight_panel, _overlay_series, sized_weights}` (the overlay engine + the +live per-coin weights), `bybit_liquidity_sweep.{_metrics, _window}` (metrics + windowing), +`bybit_liquidity.liquid_universe` (the `--min-dollar-vol` path the live book uses). READ-ONLY: only panel +READS; no PaperRepo, no `store.write_*` (a WriteTripwire test proves it). +""" +from __future__ import annotations + +import math +import statistics as st +from typing import Any, Protocol + +from fxhnt.application.bybit_liquidity_sweep import _metrics, _window +from fxhnt.application.spread_overlay import ( + _K_GRID, + _LAM_GRID, + _overlay_series, + per_coin_weight_panel, + sized_weights, +) + +_TAKER_FEE_BPS = 5.5 +_LIVE_BOUND = 10_000_000.0 +# The broader universes to consider for (B). Each is selected-against on TRAIN; the per-fold winner is tested. +_DEFAULT_BROAD_BOUNDS: tuple[float, ...] = (5_000_000.0, 1_000_000.0, 500_000.0) +# Grids INCLUDE the no-overlay point (0.0) so "broad without overlay" is an honest candidate the train can pick. +_K_GRID_WF: tuple[float, ...] = (0.0,) + _K_GRID +_LAM_GRID_WF: tuple[float, ...] = (0.0,) + _LAM_GRID + +# A coin can absorb this fraction of its median daily dollar-volume without undue impact (capacity assumption). +# The capacity RATIO (B/A) is invariant to this value, so the verdict does not depend on it. +_PARTICIPATION_FRAC = 0.10 + +# Verdict bars. +_SHARPE_UPGRADE_BAR = 0.20 # mean OOS NET Sharpe lift (B−A) to call broad+overlay meaningfully better. +_WIN_FRAC_BAR = 0.60 # fraction of folds B must beat A to count as consistent (not a one-fold fluke). +_CAPACITY_FLOOR_FRAC = 0.50 # B's deployable capital must be >= this × A's to be "acceptable". + + +class _FeatureStore(Protocol): + def crypto_close_panel(self, *, suffix: str = "USDT") -> dict[str, dict[int, float]]: ... + def read_panel(self, symbols: list[str], feature: str) -> dict[str, dict[int, float]]: ... + + +# --- walk-forward fold construction (no leakage) ------------------------------------------------ + +def walk_forward_folds(days: list[int], n_folds: int) -> list[dict[str, int]]: + """EXPANDING-WINDOW walk-forward folds over the sorted day axis. The axis is cut into `n_folds + 1` + contiguous index blocks; fold i TRAINS on days up to the end of block i (`train_hi`, inclusive) and TESTS + on the disjoint next block `[test_lo, test_hi]` (inclusive). Guarantees: every fold's train strictly + precedes its test (`train_hi < test_lo`) and the test blocks are disjoint (no leakage). Returns + `[{fold, train_hi, test_lo, test_hi}]`. Empty when there are too few days for `n_folds + 1` non-empty + blocks of >= 2 days each.""" + uniq = sorted(set(days)) + n = len(uniq) + if n_folds < 1 or n < 2 * (n_folds + 1): + return [] + block = n // (n_folds + 1) + folds: list[dict[str, int]] = [] + for i in range(n_folds): + train_end = block * (i + 1) # index one-past the end of the training span + test_end = block * (i + 2) if i < n_folds - 1 else n + if train_end < 1 or test_end - train_end < 2: + continue + folds.append({ + "fold": i + 1, + "train_hi": uniq[train_end - 1], + "test_lo": uniq[train_end], + "test_hi": uniq[test_end - 1], + }) + return folds + + +# --- one (universe, k, λ) evaluated on a [day_lo, day_hi] window --------------------------------- + +def _eval_window(daily: dict[int, dict[str, tuple[float, float]]], cost_by_coin: dict[str, float], + spread_by_coin: dict[str, float], *, k: float, lam: float, + day_lo: int | None, day_hi: int | None) -> dict[str, Any]: + """NET metrics for one overlay variant restricted to a day window. The overlay walk runs on ONLY the + window's days (held state starts fresh at the window edge — correct for a disjoint OOS test block, and + leakage-free for train selection). Returns Sharpe/total_return/max_dd + window turnover + cost-drag + the + NET series + day count.""" + sub = _window(daily, day_lo=day_lo, day_hi=day_hi) # type: ignore[arg-type] + gross, net, turnover, cost_drag = _overlay_series(sub, cost_by_coin, spread_by_coin, k=k, lam=lam) + m = _metrics(net) + return {"sharpe": m["sharpe"], "total_return": m["total_return"], "max_dd": m["max_dd"], + "turnover": turnover, "cost_drag_bps": cost_drag, "net_series": net, "days": m["days"]} + + +# --- capacity (bottleneck deployable capital) --------------------------------------------------- + +def _median_dollar_vol(store: _FeatureStore, coins: list[str], day_lo: int, day_hi: int) -> dict[str, float]: + """Median daily dollar-volume (turnover) per coin over `[day_lo, day_hi]`. Memory-bounded: reads only the + held coins' turnover panel. A coin with no turnover rows in the window → absent (treated as unassessable).""" + if not coins: + return {} + panel = store.read_panel(sorted(coins), "turnover") + out: dict[str, float] = {} + for c in coins: + vals = [v for d, v in panel.get(c, {}).items() if day_lo <= d <= day_hi] + if vals: + out[c] = float(st.median(vals)) + return out + + +def book_capacity(store: _FeatureStore, daily: dict[int, dict[str, tuple[float, float]]], + spread_by_coin: dict[str, float], *, k: float, day_lo: int, day_hi: int, + participation_frac: float = _PARTICIPATION_FRAC) -> dict[str, Any]: + """Estimate the book's DEPLOYABLE CAPITAL over `[day_lo, day_hi]` AFTER overlay sizing (strength `k`). + + Each coin's deployed weight is the mean |overlay-sized weight| over the window. To deploy capital C the + book needs |w_i|·C in coin i, which can absorb at most `participation_frac · ADV_i` (ADV_i = median daily + dollar-volume). So C <= participation_frac·ADV_i / |w_i| per coin; the BOTTLENECK deployable capital is the + MIN over held coins (the thinnest relative to its weight). The overlay downweighting thin coins (smaller + |w_i|) RAISES their bound → lifts capacity, so the estimate reflects the overlay. Also returns the + gross-weighted absorbable notional `Σ participation_frac·ADV_i·|w_i|` (the brief's sketch) and the weight + Herfindahl (concentration). Coins with no ADV data are skipped (unassessable, not zeroed). READ-ONLY.""" + test_days = [d for d in daily if day_lo <= d <= day_hi] + if not test_days: + return {"bottleneck_capital": 0.0, "weighted_absorbable": 0.0, "n_coins": 0, "hhi": 0.0} + acc: dict[str, float] = {} + for d in test_days: + base_w = {c: wc[0] for c, wc in daily[d].items()} + sized = sized_weights(base_w, spread_by_coin, k) + for c, w in sized.items(): + acc[c] = acc.get(c, 0.0) + abs(w) + n = len(test_days) + mean_w = {c: acc[c] / n for c in acc if acc[c] > 0.0} + if not mean_w: + return {"bottleneck_capital": 0.0, "weighted_absorbable": 0.0, "n_coins": 0, "hhi": 0.0} + adv = _median_dollar_vol(store, list(mean_w), day_lo, day_hi) + bounds = [participation_frac * adv[c] / mean_w[c] for c in mean_w if adv.get(c, 0.0) > 0.0] + bottleneck = min(bounds) if bounds else 0.0 + weighted = sum(participation_frac * adv.get(c, 0.0) * mean_w[c] for c in mean_w) + tot = sum(mean_w.values()) + hhi = sum((w / tot) ** 2 for w in mean_w.values()) if tot > 0 else 0.0 + return {"bottleneck_capital": bottleneck, "weighted_absorbable": weighted, + "n_coins": len(mean_w), "hhi": hhi} + + +# --- the verdict (pure, testable) --------------------------------------------------------------- + +def decide_verdict(*, mean_a: float, mean_b: float, win_frac: float, cap_a: float, cap_b: float) -> dict[str, Any]: + """UPGRADE / CAPACITY-TRADEOFF / OVERFIT / NO from the aggregated walk-forward economics. Pure. + + Δ = mean_b − mean_a (aggregate OOS NET Sharpe lift). `win_frac` = fraction of folds B beat A. Capacity + ratio = cap_b/cap_a (deployable capital). NO if no average lift; UPGRADE if the lift is meaningful + (>= +0.20) AND consistent (>= 60% of folds) AND capacity acceptable (>= 50% of live); CAPACITY-TRADEOFF if + the Sharpe win holds but capacity is materially lower; OVERFIT for a positive but small/inconsistent lift + (the single-split edge shrank toward liquid-61 under walk-forward).""" + d = mean_b - mean_a + if cap_a > 0.0: + cap_ratio = cap_b / cap_a + elif cap_b > 0.0: + cap_ratio = math.inf + else: + cap_ratio = 1.0 + + if d <= 0.0: + verdict = "NO" + elif d >= _SHARPE_UPGRADE_BAR and win_frac >= _WIN_FRAC_BAR: + verdict = "UPGRADE" if cap_ratio >= _CAPACITY_FLOOR_FRAC else "CAPACITY-TRADEOFF" + else: + verdict = "OVERFIT" + + return {"verdict": verdict, "d_sharpe": d, "win_frac": win_frac, + "capacity_ratio": cap_ratio, "mean_a": mean_a, "mean_b": mean_b, + "cap_a": cap_a, "cap_b": cap_b} + + +# --- correlation (robustness) ------------------------------------------------------------------- + +def _pearson(a: dict[int, float], b: dict[int, float]) -> float: + days = sorted(set(a) & set(b)) + if len(days) < 2: + return 0.0 + xs = [a[d] for d in days] + ys = [b[d] for d in days] + mx, my = st.mean(xs), st.mean(ys) + num = sum((x - mx) * (y - my) for x, y in zip(xs, ys, strict=True)) + dx = math.sqrt(sum((x - mx) ** 2 for x in xs)) + dy = math.sqrt(sum((y - my) ** 2 for y in ys)) + return (num / (dx * dy)) if dx > 0 and dy > 0 else 0.0 + + +# --- the driver --------------------------------------------------------------------------------- + +def broaden_overlay_validate(store: _FeatureStore, *, live_bound: float = _LIVE_BOUND, + broad_bounds: tuple[float, ...] = _DEFAULT_BROAD_BOUNDS, + n_folds: int = 4, k_grid: tuple[float, ...] = _K_GRID_WF, + lam_grid: tuple[float, ...] = _LAM_GRID_WF, + participation_frac: float = _PARTICIPATION_FRAC, + lookback_days: int = 90, taker_fee_bps: float = _TAKER_FEE_BPS, + sleeves: list[str] | None = None, + unlock_events: list[dict[str, Any]] | None = None) -> dict[str, Any]: + """Walk-forward head-to-head: is BROAD + overlay an UPGRADE over the live liquid-61 ($10M) book? READ-ONLY, + SHADOW eval. Builds the live and each broad per-coin panel once, cuts the day axis into expanding-window + folds, and for each fold SELECTS the broad book's (breadth, k, λ) on TRAIN then evaluates both books NET on + the disjoint TEST block. Aggregates OOS test Sharpe across folds, estimates each book's deployable capital, + correlates the two NET series, and returns the verdict. Writes NOTHING.""" + from fxhnt.application.bybit_liquidity import liquid_universe + + # 1) build the live ($10M) panel + each broad panel ONCE (memory-bounded by each universe). + live_universe = liquid_universe(store, min_dollar_vol=live_bound, lookback_days=lookback_days) + live_daily, live_cost, live_spread = per_coin_weight_panel( + store, universe=live_universe, sleeves=sleeves, unlock_events=unlock_events, + taker_fee_bps=taker_fee_bps) + if not live_daily or len(live_daily) < 4: + return {"available": False, "reason": "no usable liquid-61 ($10M) per-coin book " + "(run the Phase-A/B Bybit ingests first)"} + + broad_panels: dict[float, dict[str, Any]] = {} + for bound in broad_bounds: + universe = None if bound <= 0.0 else liquid_universe(store, min_dollar_vol=bound, + lookback_days=lookback_days) + if bound > 0.0 and not universe: + continue + daily, cost, spread = per_coin_weight_panel( + store, universe=universe, sleeves=sleeves, unlock_events=unlock_events, + taker_fee_bps=taker_fee_bps) + if daily and len(daily) >= 4: + broad_panels[bound] = {"daily": daily, "cost": cost, "spread": spread} + if not broad_panels: + return {"available": False, "reason": "no broad universe produced a usable per-coin book"} + + # 2) fold the day axis (union of all candidate universes' days). + all_days = set(live_daily) + for p in broad_panels.values(): + all_days |= set(p["daily"]) + folds = walk_forward_folds(sorted(all_days), n_folds) + if not folds: + return {"available": False, "reason": f"too few days for {n_folds} walk-forward folds"} + + # 3) per fold: select broad (breadth, k, λ) on TRAIN, evaluate both books on the disjoint TEST block. + fold_rows: list[dict[str, Any]] = [] + a_test_concat: dict[int, float] = {} + b_test_concat: dict[int, float] = {} + for f in folds: + train_hi, test_lo, test_hi = f["train_hi"], f["test_lo"], f["test_hi"] + + # (A) LIVE: liquid-61 naive eq-wt (k=0, λ=0), tested on the held-out block. + a = _eval_window(live_daily, live_cost, live_spread, k=0.0, lam=0.0, + day_lo=test_lo, day_hi=test_hi) + cap_a = book_capacity(store, live_daily, live_spread, k=0.0, day_lo=test_lo, day_hi=test_hi, + participation_frac=participation_frac) + + # (B) BROAD+overlay: select (breadth, k, λ) by best TRAIN Sharpe (no leakage), test on the block. + best_sel: dict[str, Any] | None = None + for bound, p in broad_panels.items(): + for k in k_grid: + for lam in lam_grid: + tr = _eval_window(p["daily"], p["cost"], p["spread"], k=k, lam=lam, + day_lo=None, day_hi=train_hi) + if tr["days"] < 2: + continue + if best_sel is None or tr["sharpe"] > best_sel["train_sharpe"]: + best_sel = {"bound": bound, "k": k, "lam": lam, "train_sharpe": tr["sharpe"]} + if best_sel is None: + continue + bp = broad_panels[best_sel["bound"]] + b = _eval_window(bp["daily"], bp["cost"], bp["spread"], k=best_sel["k"], lam=best_sel["lam"], + day_lo=test_lo, day_hi=test_hi) + cap_b = book_capacity(store, bp["daily"], bp["spread"], k=best_sel["k"], day_lo=test_lo, + day_hi=test_hi, participation_frac=participation_frac) + + a_test_concat.update(a["net_series"]) + b_test_concat.update(b["net_series"]) + fold_rows.append({ + "fold": f["fold"], "train_hi": train_hi, "test_lo": test_lo, "test_hi": test_hi, + "selected_bound": best_sel["bound"], "selected_k": best_sel["k"], "selected_lam": best_sel["lam"], + "live": {kk: a[kk] for kk in ("sharpe", "total_return", "max_dd", "turnover", "cost_drag_bps", "days")}, + "broad": {kk: b[kk] for kk in ("sharpe", "total_return", "max_dd", "turnover", "cost_drag_bps", "days")}, + "d_sharpe": b["sharpe"] - a["sharpe"], + "cap_live": cap_a["bottleneck_capital"], "cap_broad": cap_b["bottleneck_capital"], + "cap_live_detail": cap_a, "cap_broad_detail": cap_b, + }) + + if not fold_rows: + return {"available": False, "reason": "no fold produced a usable head-to-head"} + + # 4) aggregate across folds. + a_sharpes = [r["live"]["sharpe"] for r in fold_rows] + b_sharpes = [r["broad"]["sharpe"] for r in fold_rows] + mean_a, mean_b = st.mean(a_sharpes), st.mean(b_sharpes) + std_a = st.pstdev(a_sharpes) if len(a_sharpes) > 1 else 0.0 + std_b = st.pstdev(b_sharpes) if len(b_sharpes) > 1 else 0.0 + win_frac = sum(1 for r in fold_rows if r["d_sharpe"] > 0.0) / len(fold_rows) + # capacity comparison: aggregate (median across folds — robust to a single thin-coin spike). + cap_a_agg = float(st.median([r["cap_live"] for r in fold_rows])) + cap_b_agg = float(st.median([r["cap_broad"] for r in fold_rows])) + + verdict = decide_verdict(mean_a=mean_a, mean_b=mean_b, win_frac=win_frac, + cap_a=cap_a_agg, cap_b=cap_b_agg) + + # 5) robustness: correlation of the two OOS NET series + concentration (LARGE+FREQUENT+CONCENTRATED). + corr = _pearson(a_test_concat, b_test_concat) + hhis = [r["cap_broad_detail"]["hhi"] for r in fold_rows if r["cap_broad_detail"]["n_coins"] > 0] + broad_hhi = float(st.mean(hhis)) if hhis else 0.0 + + return { + "available": True, + "n_folds": len(fold_rows), + "live_bound": live_bound, + "broad_bounds": list(broad_panels), + "participation_frac": participation_frac, + "folds": fold_rows, + "aggregate": { + "mean_live_sharpe": mean_a, "mean_broad_sharpe": mean_b, + "std_live_sharpe": std_a, "std_broad_sharpe": std_b, + "d_sharpe": mean_b - mean_a, "win_frac": win_frac, + "cap_live": cap_a_agg, "cap_broad": cap_b_agg, + "capacity_ratio": verdict["capacity_ratio"], + "corr_broad_to_live": corr, "broad_hhi": broad_hhi, + }, + "verdict": verdict, + } diff --git a/src/fxhnt/cli.py b/src/fxhnt/cli.py index 511a885..a4362b1 100644 --- a/src/fxhnt/cli.py +++ b/src/fxhnt/cli.py @@ -3757,6 +3757,122 @@ def spread_overlay_test_cmd( "Corwin–Schultz MEASURED spread (NO assumed impact).") +@app.command("broaden-overlay-validate") +def broaden_overlay_validate_cmd( + live_bound: float = typer.Option( + 10_000_000.0, "--live-bound", + help="The LIVE book's --min-dollar-vol (the current liquid-61 deploy: $10M)."), + min_dollar_vols: str = typer.Option( + "5000000,1000000,500000", "--min-dollar-vols", + help="Comma-separated BROADER --min-dollar-vol universes to consider for (B) (0 = broad/all). Each is " + "selected-against on each fold's TRAIN; the per-fold winner is tested. DEFAULT = $5M/$1M/$500k."), + n_folds: int = typer.Option( + 4, "--n-folds", help="Number of expanding-window walk-forward TEST blocks (knobs+breadth picked on " + "TRAIN, evaluated on the disjoint TEST block)."), + participation_frac: float = typer.Option( + 0.10, "--participation-frac", + help="A coin can absorb this fraction of its median daily $-volume (capacity assumption). The capacity " + "RATIO B/A is invariant to it (it cancels), so the verdict does not depend on this value."), + sleeves: str = typer.Option( + "", "--sleeves", + help="Comma-separated book sleeves (default = the 4-edge book: tstrend,unlock,xsfunding,positioning)."), + taker_fee_bps: float = typer.Option( + 5.5, "--taker-fee-bps", + help="Bybit's PUBLISHED perp taker fee (bps). NET cost = fee + half the coin's measured CS spread."), +) -> None: + """READ-ONLY: is a BROAD universe + the spread-aware overlay a DEPLOY-BOOK UPGRADE over the current + liquid-61 ($10M) Bybit book? WALK-FORWARD head-to-head — the rigor Step 3 lacked. A SHADOW eval that NEVER + changes the live book's returns. + + Step 3 hinted broad+overlay BEAT the live book on ONE OOS split with TRAIN-tuned k/λ ($1M broad+overlay + NET-OOS Sharpe 1.30 vs $10M liquid-61 1.04) — exactly where overfitting hides. This stress-tests it: + EXPANDING-WINDOW folds; per fold the broad book's (breadth, k, λ) is SELECTED on TRAIN and BOTH books are + evaluated NET on the disjoint TEST block (train_hi < test_lo → no leakage). Aggregates OOS test Sharpe + across folds, estimates each book's DEPLOYABLE CAPITAL (bottleneck coin: participation_frac·ADV / |weight|; + the overlay downweighting thin coins lifts it), and correlates the two NET series. + + VERDICT: UPGRADE (aggregate lift >= +0.20, broad wins a majority of folds, capacity >= 50% of live — wire + it in) / CAPACITY-TRADEOFF (Sharpe win but materially lower capacity — a choice) / OVERFIT (the 1.30 + shrinks toward liquid-61 under walk-forward — keep liquid-61) / NO. Writes NOTHING.""" + import logging + + from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore + from fxhnt.application.broaden_overlay_validate import broaden_overlay_validate + + log = logging.getLogger(__name__) + settings = get_settings() + store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features") + bounds = tuple(float(x) for x in min_dollar_vols.split(",") if x.strip()) + sleeve_list = [s.strip() for s in sleeves.split(",") if s.strip()] or None + + def _load_unlock_events() -> list: # type: ignore[type-arg] + from fxhnt.adapters.orchestration.assets import _data_dir + from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo + try: + return load_unlock_events(operational_unlock_repo(settings), + f"{_data_dir()}/unlock_calendar.json") + except Exception as exc: # noqa: BLE001 — absent calendar → unlock edge omitted, not a crash + log.warning("broaden-overlay-validate: could not load unlock calendar: %s", exc) + return [] + + try: + unlock_events = _load_unlock_events() + rep = broaden_overlay_validate(store, live_bound=live_bound, broad_bounds=bounds, n_folds=n_folds, + participation_frac=participation_frac, taker_fee_bps=taker_fee_bps, + sleeves=sleeve_list, unlock_events=unlock_events) + finally: + store.close() + + typer.echo("\n=== BROADEN + OVERLAY deploy-book validation — WALK-FORWARD head-to-head vs liquid-61 — " + "READ-ONLY — SHADOW eval ===") + if not rep.get("available"): + typer.echo(f"INSUFFICIENT DATA: {rep.get('reason', 'no usable book')}") + typer.echo(" (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` first — the high/low + " + "turnover features are needed for the measured cost + capacity model)") + return + + typer.echo(f"\nWalk-forward: {rep['n_folds']} expanding-window folds. (A) LIVE = liquid-61 " + f"(${rep['live_bound']/1e6:.0f}M, naive eq-wt). (B) BROAD+overlay = per-fold-train-selected " + f"breadth (from {[f'${b/1e6:g}M' for b in rep['broad_bounds']]}) + spread overlay.") + typer.echo("\n fold window(test) (A) live Sh (B) broad Sh Δ Sh sel breadth/k/λ " + "cap live cap broad") + typer.echo(" " + "-" * 116) + for f in rep["folds"]: + sel = f"${f['selected_bound']/1e6:g}M k={f['selected_k']:g} λ={f['selected_lam']:g}" + typer.echo(f" {f['fold']:>4d} [{f['test_lo']//86400},{f['test_hi']//86400}]" + f"{'':<6} {f['live']['sharpe']:>10.2f} {f['broad']['sharpe']:>11.2f} " + f"{f['d_sharpe']:>+6.2f} {sel:<22} ${f['cap_live']:>11,.0f} ${f['cap_broad']:>11,.0f}") + + agg = rep["aggregate"] + typer.echo(" " + "-" * 116) + typer.echo(f" AGG mean OOS NET Sh live {agg['mean_live_sharpe']:>+.2f}±{agg['std_live_sharpe']:.2f}" + f" broad {agg['mean_broad_sharpe']:>+.2f}±{agg['std_broad_sharpe']:.2f} " + f"Δ {agg['d_sharpe']:>+.2f} broad-wins {agg['win_frac']*100:.0f}% of folds") + typer.echo(f"\n CAPACITY (deployable, median across folds): live ${agg['cap_live']:,.0f} vs " + f"broad ${agg['cap_broad']:,.0f} (ratio {agg['capacity_ratio']:.2f}×)") + typer.echo(f" ROBUSTNESS: corr(broad NET, live NET) = {agg['corr_broad_to_live']:+.2f} " + f"(1 = same-book; <1 = genuinely different). broad weight HHI = {agg['broad_hhi']:.3f} " + f"(concentration).") + + v = rep["verdict"] + headline = { + "UPGRADE": "UPGRADE — broad+overlay's aggregate walk-forward NET Sharpe is meaningfully higher AND " + "capacity is acceptable. WIRE IT IN.", + "CAPACITY-TRADEOFF": "CAPACITY-TRADEOFF — broad+overlay wins Sharpe but holds materially less " + "deployable capital. A choice, not a free win.", + "OVERFIT": "OVERFIT — the single-split lift shrinks toward liquid-61 under walk-forward (Step 3 was a " + "fluke). KEEP liquid-61.", + "NO": "NO — broad+overlay is no better on average. KEEP liquid-61.", + }[v["verdict"]] + typer.echo(f"\nVERDICT: {headline}") + typer.echo(f" (Δ aggregate OOS NET Sharpe {v['d_sharpe']:+.2f}; broad wins {v['win_frac']*100:.0f}% of " + f"folds; capacity ratio {v['capacity_ratio']:.2f}×.)") + typer.echo("\nNOTE: writes NOTHING — the live book is untouched (SHADOW eval). Per-fold knobs AND breadth " + "are selected on TRAIN; the verdict uses the disjoint held-out TEST blocks (no leakage). " + "Capacity = bottleneck participation_frac·ADV/|weight|; the ratio is invariant to " + "participation_frac. Cost = Bybit's 5.5bp taker + half the Corwin–Schultz MEASURED spread.") + + @app.command("bybit-drawdown-kill-test") def bybit_drawdown_kill_test( train_frac: float = typer.Option( diff --git a/tests/integration/test_broaden_overlay_validate.py b/tests/integration/test_broaden_overlay_validate.py new file mode 100644 index 0000000..c09b1bc --- /dev/null +++ b/tests/integration/test_broaden_overlay_validate.py @@ -0,0 +1,297 @@ +"""WALK-FORWARD validation of the deploy-book question: is a BROAD universe + spread-aware overlay an UPGRADE +over the current liquid-61 ($10M) Bybit book? READ-ONLY, SHADOW eval (never mutates the live book). + +Step 3 (`spread_overlay`) hinted broad+overlay BEAT the live book on ONE OOS split with TRAIN-tuned k/λ — the +exact place overfitting hides. This module re-tests it with WALK-FORWARD folds + a capacity check, so the +verdict is UPGRADE / CAPACITY-TRADEOFF / OVERFIT / NO. + +Tests (TDD): + * walk-forward splits are correct — train strictly precedes test (no leakage), test blocks disjoint; + * baseline (k=0, λ=0, full window) reconciles to the existing naive per-coin book (no engine drift); + * capacity behaves — thinner coins → lower deployable capital; overlay downweighting a thin coin RAISES it; + * the verdict reacts to economics (NOT hard-coded): broad wins across folds → UPGRADE; broad wins one fold + only → OVERFIT; broad wins Sharpe but tanks capacity → CAPACITY-TRADEOFF; no lift → NO; + * the driver runs on a seeded store with disjoint folds + a head-to-head per fold + an aggregate verdict; + * READ-ONLY (WriteTripwire) + determinism + CLI smoke. +NO network — the store is in-memory SQLite; external inputs are injected. +""" +from __future__ import annotations + +import math + +from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore +from fxhnt.application.broaden_overlay_validate import ( + _eval_window, + book_capacity, + broaden_overlay_validate, + decide_verdict, + walk_forward_folds, +) +from fxhnt.application.bybit_liquidity_sweep import per_coin_book_returns +from fxhnt.application.spread_overlay import per_coin_weight_panel + +_DAY = 86_400 + + +# --- fixtures (mirror the spread-overlay / liquidity-sweep fixtures) ------------------------------ + +def _ohlc_from_close(close: float, spread_frac: float, vol_frac: float, phase: float) -> tuple[float, float]: + vol = vol_frac * (0.5 + 0.5 * math.cos(7.0 * phase)) + half = 0.5 * spread_frac + vol + return close * (1.0 + half), close * (1.0 - half) + + +def _seed_carry(store: TimescaleFeatureStore, funding: dict[str, float], spread_by_sym: dict[str, float], *, + days: int = 600) -> None: + for idx, (sym, fund) in enumerate(funding.items()): + sf = spread_by_sym[sym] + rows = [] + for d in range(days): + f = fund + 0.0003 * math.cos(0.5 * d + idx) + px = 100.0 + high, low = _ohlc_from_close(px, sf, vol_frac=0.0005, phase=0.5 * d + idx) + rows.append((d * _DAY, {"funding": f, "close": px, "spot_close": px, "high": high, "low": low})) + store.write_features(sym, rows) + + +def _seed_turnover(store: TimescaleFeatureStore, dv: dict[str, float], *, days: int = 600) -> None: + for sym, v in dv.items(): + store.write_features(sym, [(d * _DAY, {"turnover": v}) for d in range(days)]) + + +def _seed_two_tier(store: TimescaleFeatureStore, *, days: int = 600) -> None: + """A deep tier (tight spread, high $vol — clears $10M) + a thin tier (wide spread, low $vol — clears $1M + but fails $10M). The broad universe pulls in the thin coins the overlay can help with.""" + deep = {"AAAUSDT": 0.001, "BBBUSDT": 0.001, "CCCUSDT": 0.001, + "XXXUSDT": -0.001, "YYYUSDT": -0.001, "ZZZUSDT": -0.001} + thin = {"PPPUSDT": 0.0012, "QQQUSDT": 0.0012, "RRRUSDT": 0.0012, + "SSSUSDT": -0.0012, "TTTUSDT": -0.0012, "UUUUSDT": -0.0012} + spread = {s: 0.0004 for s in deep} + spread.update({s: 0.02 for s in thin}) # 200bp measured spread on the thin tier + _seed_carry(store, {**deep, **thin}, spread, days=days) + dv = {s: 100_000_000.0 for s in deep} + dv.update({s: 2_000_000.0 for s in thin}) # thin coins clear $1M, fail $10M + _seed_turnover(store, dv, days=days) + + +class _WriteTripwireStore: + _FORBIDDEN = frozenset({ + "write_features", "write_features_bulk", "upsert_feature_rows", "upsert_membership", + "replace_positions", "upsert_nav", "upsert_sleeve_ret", "replace_shadow_positions", + "replace_trades", "_create_schema", "_bulk_upsert", + }) + + def __init__(self, inner: TimescaleFeatureStore) -> None: + object.__setattr__(self, "_inner", inner) + + def __getattr__(self, name: str): + if name in _WriteTripwireStore._FORBIDDEN: + raise AssertionError(f"READ-ONLY violation: broaden-overlay-validate called write method {name!r}") + return getattr(self._inner, name) + + +# --- walk-forward fold correctness (no leakage) -------------------------------------------------- + +def test_walk_forward_folds_no_leakage_and_disjoint() -> None: + days = list(range(0, 100)) + folds = walk_forward_folds(days, n_folds=4) + assert len(folds) == 4 + prev_test_hi = -1 + for f in folds: + # train strictly precedes test (no leakage). + assert f["train_hi"] < f["test_lo"], f + assert f["test_lo"] <= f["test_hi"] + # test blocks are disjoint and ordered (each begins after the previous one ended). + assert f["test_lo"] > prev_test_hi, f + prev_test_hi = f["test_hi"] + + +def test_walk_forward_folds_expands_train() -> None: + """Expanding window: each fold's train_hi is >= the previous fold's (the train set grows).""" + folds = walk_forward_folds(list(range(0, 120)), n_folds=3) + train_his = [f["train_hi"] for f in folds] + assert train_his == sorted(train_his) + assert len(set(train_his)) == len(train_his) # strictly growing + + +def test_walk_forward_folds_too_few_days_returns_empty() -> None: + assert walk_forward_folds([1, 2, 3], n_folds=4) == [] + assert walk_forward_folds(list(range(20)), n_folds=0) == [] + + +# --- baseline reconciliation --------------------------------------------------------------------- + +def test_baseline_window_reconciles_to_naive_book() -> None: + """k=0, λ=0 over the FULL window must reproduce the naive eq-wt per-coin book EXACTLY (no engine drift).""" + store = TimescaleFeatureStore("sqlite://", table="bybit_features") + funding = {"AAAUSDT": 0.001, "BBBUSDT": 0.001, "XXXUSDT": -0.001, "YYYUSDT": -0.001} + spread = {s: 0.0008 for s in funding} + _seed_carry(store, funding, spread, days=200) + _seed_turnover(store, {s: 50_000_000.0 for s in funding}, days=200) + daily, cost, spr = per_coin_weight_panel(store, universe=None, sleeves=["xsfunding"], unlock_events=[]) + ev = _eval_window(daily, cost, spr, k=0.0, lam=0.0, day_lo=None, day_hi=None) + pc = per_coin_book_returns(store, universe=None, sleeves=["xsfunding"], unlock_events=[]) + store.close() + common = sorted(set(ev["net_series"]) & set(pc["net_series"])) + assert common + for d in common: + assert abs(ev["net_series"][d] - pc["net_series"][d]) < 1e-12, (d, ev["net_series"][d], + pc["net_series"][d]) + + +# --- capacity behaviour -------------------------------------------------------------------------- + +def _hand_panel(weights: dict[str, float], n_days: int = 40) -> dict[int, dict[str, tuple[float, float]]]: + """A constant-weight panel (per-unit return 0 — capacity is about weights/volume, not P&L).""" + return {d: {c: (w, 0.0) for c, w in weights.items()} for d in range(n_days)} + + +def test_capacity_thinner_coins_lower_capital() -> None: + """A book that must hold a THIN coin has lower deployable capital than one holding only a DEEP coin.""" + store = TimescaleFeatureStore("sqlite://", table="bybit_features") + _seed_turnover(store, {"DEEPUSDT": 100_000_000.0, "THINUSDT": 1_000_000.0}, days=40) + deep_only = book_capacity(store, _hand_panel({"DEEPUSDT": 1.0}), {"DEEPUSDT": 0.0}, + k=0.0, day_lo=0, day_hi=39 * _DAY) + with_thin = book_capacity(store, _hand_panel({"DEEPUSDT": 0.5, "THINUSDT": 0.5}), + {"DEEPUSDT": 0.0, "THINUSDT": 0.0}, k=0.0, day_lo=0, day_hi=39 * _DAY) + store.close() + assert with_thin["bottleneck_capital"] < deep_only["bottleneck_capital"], (with_thin, deep_only) + + +def test_capacity_overlay_downweighting_raises_capital() -> None: + """Overlay sizing (k>0) downweights the WIDE-spread thin coin → raises its absorbable bound → lifts the + book's bottleneck deployable capital vs no overlay (k=0).""" + store = TimescaleFeatureStore("sqlite://", table="bybit_features") + _seed_turnover(store, {"DEEPUSDT": 100_000_000.0, "THINUSDT": 1_000_000.0}, days=40) + panel = _hand_panel({"DEEPUSDT": 0.5, "THINUSDT": 0.5}) + spread = {"DEEPUSDT": 1.0, "THINUSDT": 500.0} # thin coin has a wide measured spread + no_overlay = book_capacity(store, panel, spread, k=0.0, day_lo=0, day_hi=39 * _DAY) + with_overlay = book_capacity(store, panel, spread, k=0.05, day_lo=0, day_hi=39 * _DAY) + store.close() + assert with_overlay["bottleneck_capital"] > no_overlay["bottleneck_capital"], (with_overlay, no_overlay) + + +def test_capacity_ratio_invariant_to_participation_frac() -> None: + """The capacity RATIO (B/A) is invariant to participation_frac (it cancels), so the verdict does not depend + on that assumption.""" + store = TimescaleFeatureStore("sqlite://", table="bybit_features") + _seed_turnover(store, {"DEEPUSDT": 100_000_000.0, "THINUSDT": 1_000_000.0}, days=40) + panel = _hand_panel({"DEEPUSDT": 0.5, "THINUSDT": 0.5}) + spread = {"DEEPUSDT": 0.0, "THINUSDT": 0.0} + a = book_capacity(store, panel, spread, k=0.0, day_lo=0, day_hi=39 * _DAY, participation_frac=0.05) + b = book_capacity(store, panel, spread, k=0.0, day_lo=0, day_hi=39 * _DAY, participation_frac=0.20) + store.close() + assert b["bottleneck_capital"] == 4.0 * a["bottleneck_capital"] or abs( + b["bottleneck_capital"] - 4.0 * a["bottleneck_capital"]) < 1e-6 + + +# --- the verdict reacts to economics (NOT hard-coded) -------------------------------------------- + +def test_verdict_upgrade_when_broad_wins_across_folds() -> None: + v = decide_verdict(mean_a=1.0, mean_b=1.4, win_frac=1.0, cap_a=1_000_000.0, cap_b=900_000.0) + assert v["verdict"] == "UPGRADE", v + + +def test_verdict_overfit_when_broad_wins_only_one_fold() -> None: + """A big win on a single fold (mean lift looks positive) but inconsistent across folds → OVERFIT, not + UPGRADE — the single-split edge did not survive walk-forward.""" + v = decide_verdict(mean_a=1.0, mean_b=1.3, win_frac=0.25, cap_a=1_000_000.0, cap_b=1_000_000.0) + assert v["verdict"] == "OVERFIT", v + + +def test_verdict_capacity_tradeoff_when_sharpe_wins_but_capacity_tanks() -> None: + v = decide_verdict(mean_a=1.0, mean_b=1.4, win_frac=1.0, cap_a=1_000_000.0, cap_b=100_000.0) + assert v["verdict"] == "CAPACITY-TRADEOFF", v + + +def test_verdict_no_when_no_lift() -> None: + v = decide_verdict(mean_a=1.2, mean_b=1.1, win_frac=0.5, cap_a=1_000_000.0, cap_b=1_000_000.0) + assert v["verdict"] == "NO", v + + +def test_verdict_overfit_when_lift_small_but_consistent() -> None: + """A small (<0.20) but consistent lift still shrinks toward liquid-61 → OVERFIT (not a meaningful upgrade).""" + v = decide_verdict(mean_a=1.0, mean_b=1.1, win_frac=1.0, cap_a=1_000_000.0, cap_b=1_000_000.0) + assert v["verdict"] == "OVERFIT", v + + +# --- the driver (walk-forward head-to-head on a seeded store) ------------------------------------ + +def test_driver_runs_walk_forward_head_to_head() -> None: + store = TimescaleFeatureStore("sqlite://", table="bybit_features") + _seed_two_tier(store) + rep = broaden_overlay_validate(store, broad_bounds=(1_000_000.0,), n_folds=3, + sleeves=["xsfunding"], unlock_events=[]) + store.close() + assert rep["available"] is True + assert rep["n_folds"] >= 2 + # disjoint test windows + train precedes test on every fold (no leakage). + prev_hi = -1 + for f in rep["folds"]: + assert f["train_hi"] < f["test_lo"] + assert f["test_lo"] > prev_hi + prev_hi = f["test_hi"] + # both books reported per fold. + assert math.isfinite(f["live"]["sharpe"]) + assert math.isfinite(f["broad"]["sharpe"]) + assert f["cap_live"] >= 0.0 and f["cap_broad"] >= 0.0 + agg = rep["aggregate"] + assert math.isfinite(agg["mean_live_sharpe"]) and math.isfinite(agg["mean_broad_sharpe"]) + assert 0.0 <= agg["win_frac"] <= 1.0 + assert rep["verdict"]["verdict"] in ("UPGRADE", "CAPACITY-TRADEOFF", "OVERFIT", "NO") + + +def test_driver_is_read_only() -> None: + inner = TimescaleFeatureStore("sqlite://", table="bybit_features") + _seed_two_tier(inner) + guarded = _WriteTripwireStore(inner) + rep = broaden_overlay_validate(guarded, broad_bounds=(1_000_000.0,), n_folds=3, + sleeves=["xsfunding"], unlock_events=[]) + inner.close() + assert rep["available"] is True + + +def test_driver_is_deterministic() -> None: + store = TimescaleFeatureStore("sqlite://", table="bybit_features") + _seed_two_tier(store) + a = broaden_overlay_validate(store, broad_bounds=(1_000_000.0,), n_folds=3, + sleeves=["xsfunding"], unlock_events=[]) + b = broaden_overlay_validate(store, broad_bounds=(1_000_000.0,), n_folds=3, + sleeves=["xsfunding"], unlock_events=[]) + store.close() + assert a["aggregate"] == b["aggregate"] + assert a["verdict"] == b["verdict"] + + +def test_driver_empty_store_unavailable() -> None: + store = TimescaleFeatureStore("sqlite://", table="bybit_features") + rep = broaden_overlay_validate(store, broad_bounds=(1_000_000.0,), n_folds=3, + sleeves=["xsfunding"], unlock_events=[]) + store.close() + assert rep["available"] is False + + +# --- CLI smoke ----------------------------------------------------------------------------------- + +def test_cli_broaden_overlay_validate_prints(monkeypatch) -> None: + from typer.testing import CliRunner + + import fxhnt.cli as cli + + seeded = TimescaleFeatureStore("sqlite://", table="bybit_features") + _seed_two_tier(seeded) + + monkeypatch.setattr( + "fxhnt.adapters.warehouse.timescale_feature_store.TimescaleFeatureStore", + lambda *a, **k: seeded, raising=False) + + result = CliRunner().invoke( + cli.app, ["broaden-overlay-validate", "--min-dollar-vols", "1000000", + "--n-folds", "3", "--sleeves", "xsfunding"]) + seeded.close() + + assert result.exit_code == 0, result.output + out = result.output.lower() + assert "fold" in out + assert "verdict" in out + assert "capacity" in out