From d231971cb832ba328866eceafcd82758a25e3012 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Thu, 18 Jun 2026 23:33:13 +0200 Subject: [PATCH] feat(xsfunding): gauntlet verdict matrix (modes x floors, crypto 365d, cross-cell DSR n_trials) Co-Authored-By: Claude Opus 4.8 (1M context) --- .../application/funding_backtest_runner.py | 55 +++++++++++++++++++ .../test_funding_backtest_runner.py | 20 ++++++- 2 files changed, 74 insertions(+), 1 deletion(-) diff --git a/src/fxhnt/application/funding_backtest_runner.py b/src/fxhnt/application/funding_backtest_runner.py index 1d0e633..e093fbb 100644 --- a/src/fxhnt/application/funding_backtest_runner.py +++ b/src/fxhnt/application/funding_backtest_runner.py @@ -4,12 +4,17 @@ Produces a daily net-return series per construction mode.""" from __future__ import annotations from dataclasses import dataclass +from typing import Any import numpy as np +from fxhnt.domain.backtest import compute_stats from fxhnt.domain.cross_sectional_funding import ( Panel, construction_weights, eligible_asof, funding_score, ) +from fxhnt.domain.gauntlet.core import evaluate, per_period_sharpe + +_PERIODS_PER_YEAR = 365 # crypto trades daily, all year _MODES = ("long_tilt", "market_neutral", "executable") @@ -84,3 +89,53 @@ def _book(w: dict[str, float], prev_w: dict[str, float], turnover = sum(abs(w.get(s, 0.0) - prev_w.get(s, 0.0)) * cost.get(s, 0.0) for s in set(w) | set(prev_w)) / 2.0 return realized - turnover + + +def _json_floats(d: dict[str, Any]) -> dict[str, Any]: + """Coerce model_dump scalars to JSON-native: numpy → python float, NaN/inf → 0.0.""" + out: dict[str, Any] = {} + for k, v in d.items(): + if isinstance(v, (np.floating, float)): + fv = float(v) + out[k] = fv if np.isfinite(fv) else 0.0 + elif isinstance(v, (np.integer,)): + out[k] = int(v) + else: + out[k] = v + return out + + +def evaluate_funding_matrix(panel: Panel, *, floors: list[float], lookback_days: int = 7, + quantile: float = 0.2, cost_bps: float = 8.0, slip_coef: float = 0.0005, + oos_fraction: float = 0.40, dsr_min: float = 0.95, + oos_min_sharpe: float = 0.0, max_is_oos_decay: float = 0.50) -> dict[str, Any]: + """Run the runner at each liquidity floor, gauntlet every (mode, floor) cell. + n_trials = total cells tested (multiple-testing correction).""" + runs = {f: FundingBacktestRunner(panel, min_qvol=f, lookback_days=lookback_days, quantile=quantile, + cost_bps=cost_bps, slip_coef=slip_coef).run() for f in floors} + series_by_cell = {(m, f): runs[f].returns_by_mode[m] for f in floors for m in _MODES} + n_trials = len(series_by_cell) + is_sharpes = [per_period_sharpe(r[:int((1.0 - oos_fraction) * len(r))]) + for r in series_by_cell.values() if len(r) > 3] + sr_variance = float(np.var(is_sharpes)) if len(is_sharpes) > 1 else 0.0 + out: dict[str, Any] = {"cells": {}, "n_trials": n_trials, + "settings": {"floors": floors, "lookback_days": lookback_days, + "quantile": quantile, "cost_bps": cost_bps, + "slip_coef": slip_coef, "oos_fraction": oos_fraction}} + for (m, f), r in series_by_cell.items(): + stats = compute_stats(r, periods_per_year=_PERIODS_PER_YEAR) + key = f"{m}@{f:.0e}" + if len(r) < 3: + out["cells"][key] = {"mode": m, "floor": float(f), "stats": _json_floats(stats.model_dump()), + "verdict": {"passed": False, "dsr": 0.0, "is_sharpe": 0.0, + "oos_sharpe": 0.0, "n_trials": n_trials, + "reasons": ["insufficient data"], "pvalue": 1.0}} + continue + split = int((1.0 - oos_fraction) * len(r)) + v = evaluate(r[:split], r[split:], n_trials=n_trials, sr_variance=sr_variance, + dsr_min=dsr_min, oos_min_sharpe=oos_min_sharpe, max_is_oos_decay=max_is_oos_decay, + has_economic_rationale=True) # funding carry is a documented structural premium + out["cells"][key] = {"mode": m, "floor": float(f), + "stats": _json_floats(stats.model_dump()), + "verdict": _json_floats(v.model_dump())} + return out diff --git a/tests/integration/test_funding_backtest_runner.py b/tests/integration/test_funding_backtest_runner.py index 8c2c399..51eb91e 100644 --- a/tests/integration/test_funding_backtest_runner.py +++ b/tests/integration/test_funding_backtest_runner.py @@ -1,5 +1,9 @@ import numpy as np -from fxhnt.application.funding_backtest_runner import FundingBacktestRunner, _book +from fxhnt.application.funding_backtest_runner import ( + FundingBacktestRunner, + _book, + evaluate_funding_matrix, +) def _panel(): @@ -103,3 +107,17 @@ def test_book_carry_and_turnover(): r3 = _book({"B": 1.0}, {"A": 1.0}, {"B": 0.0}, {"A": 0.002, "B": 0.002}) expected_turn = (abs(1.0) * 0.002 + abs(-1.0) * 0.002) / 2.0 # = 0.002 assert abs(r3 - (0.0 - expected_turn)) < 1e-12 + + +def test_evaluate_matrix_runs_all_cells(): + panel = _panel() # the existing helper in this test file (HI/MID/LO/NEG, 200 days) + report = evaluate_funding_matrix(panel, floors=[1e6, 5e6], lookback_days=7, quantile=0.5, + cost_bps=8.0, slip_coef=0.0, oos_fraction=0.4) + assert len(report["cells"]) == 6 # 3 modes x 2 floors + for cell in report["cells"].values(): + assert {"mode", "floor", "stats", "verdict"} <= set(cell) + assert {"sharpe", "cagr", "max_drawdown", "n_obs"} <= set(cell["stats"]) + assert {"passed", "dsr", "is_sharpe", "oos_sharpe", "n_trials"} <= set(cell["verdict"]) + assert cell["verdict"]["n_trials"] == 6 # multiple-testing across the whole matrix + import json + json.dumps(report) # must be JSON-serializable (no NaN/np types)