feat(xsfunding): gauntlet verdict matrix (modes x floors, crypto 365d, cross-cell DSR n_trials)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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

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)