Merge: broaden+overlay deploy-book validation (walk-forward + capacity vs liquid-61)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
369
src/fxhnt/application/broaden_overlay_validate.py
Normal file
369
src/fxhnt/application/broaden_overlay_validate.py
Normal file
@@ -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,
|
||||
}
|
||||
116
src/fxhnt/cli.py
116
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(
|
||||
|
||||
297
tests/integration/test_broaden_overlay_validate.py
Normal file
297
tests/integration/test_broaden_overlay_validate.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user