From 549a21f660cff1cd1723b5092e38170efab672a1 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Thu, 18 Jun 2026 22:58:52 +0200 Subject: [PATCH] =?UTF-8?q?docs(xsfunding):=20cross-sectional=20funding-di?= =?UTF-8?q?spersion=20implementation=20plan=20(6=20tasks=20=E2=86=92=20ver?= =?UTF-8?q?dict=20matrix)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 (1M context) --- .../2026-06-18-cross-sectional-funding.md | 694 ++++++++++++++++++ 1 file changed, 694 insertions(+) create mode 100644 docs/superpowers/plans/2026-06-18-cross-sectional-funding.md diff --git a/docs/superpowers/plans/2026-06-18-cross-sectional-funding.md b/docs/superpowers/plans/2026-06-18-cross-sectional-funding.md new file mode 100644 index 0000000..ba1f559 --- /dev/null +++ b/docs/superpowers/plans/2026-06-18-cross-sectional-funding.md @@ -0,0 +1,694 @@ +# Cross-Sectional Crypto Funding-Dispersion — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax. + +**Goal:** Backtest (survivorship-free, PIT) a cross-sectional crypto funding-dispersion strategy across a matrix of constructions × liquidity floors, through the gauntlet, to find a real uncrowded small-capital edge — then surface the verdict. + +**Architecture:** Pure domain (PIT eligibility + cross-sectional funding score + 3 construction modes + signed delta-neutral carry book-return) consumed by a walk-forward runner that reads the existing survivorship-free `crypto_pit` npz panel, producing a per-(mode × floor) daily return series, each scored by the existing gauntlet (DSR/OOS). A CLI runs it and prints the verdict matrix. + +**Tech Stack:** Python 3.12, numpy, the `crypto_pit` npz panel (foxhunt `data/surfer/crypto_pit/`, 180 symbols incl. 25 delisted, 2019–2026), the gauntlet (`compute_stats`/`evaluate`), Typer CLI, pytest (`.venv/bin/python -m pytest`), strict mypy. + +**Spec:** `docs/superpowers/specs/2026-06-18-cross-sectional-funding-design.md`. + +**House rules:** TDD (failing test first). Pure functions in `domain/`. Synthetic in-memory panels in tests — NO live network, `.npz` loaded WITHOUT `allow_pickle`. Strict mypy (no NEW errors). Commit per task, do NOT push. Trailer: +``` +Co-Authored-By: Claude Opus 4.8 (1M context) +``` +**Crypto convention:** daily series, **365 periods/year** (pass `periods_per_year=365` to `compute_stats`). + +**Scope:** data adapter → domain → runner → gauntlet verdict matrix → CLI. Paper-forward wiring of the *winning* construction is a deliberate follow-on AFTER we see the verdict (don't wire a possibly-falsified strategy into the cockpit) — NOT in this plan. + +--- + +## Reuse contract (exact — verified this session) +```python +# src/fxhnt/domain/strategies/equity_factor.py +def robust_z(values: list[float | None], k: float = 3.0) -> list[float] # median/MAD winsorized z, missing→0.0 + +# src/fxhnt/domain/backtest.py +class BacktestStats(BaseModel): cagr: float; ann_vol: float; sharpe: float; max_drawdown: float; n_obs: int +def compute_stats(returns: np.ndarray, periods_per_year: int = 252) -> BacktestStats + +# src/fxhnt/domain/gauntlet/core.py +def evaluate(is_returns, oos_returns, n_trials, sr_variance, *, dsr_min=0.95, oos_min_sharpe=0.0, + max_is_oos_decay=0.50, has_economic_rationale=None) -> Verdict +def per_period_sharpe(returns: np.ndarray) -> float # per discovery.py; if the real name differs, use what discovery.py uses +class Verdict(BaseModel): passed; dsr; is_sharpe; oos_sharpe; n_trials; reasons; pvalue + +# src/fxhnt/config.py — Settings.gauntlet has oos_fraction/dsr_min/oos_min_sharpe/max_is_oos_decay +``` +**npz panel format** (verified): each `data/surfer/crypto_pit/.npz` has parallel arrays `day` (int epoch-day), `close`, `qvol` (daily USD quote-volume), `funding` (daily-summed 8h funding rate). Symbol = filename stem. + +--- + +## File Structure +- Create `src/fxhnt/adapters/data/crypto_pit_panel.py` — npz loader → in-memory panel. +- Create `src/fxhnt/domain/cross_sectional_funding.py` — pure: `eligible_asof`, `funding_score`, `construction_weights`, `book_return_xs`, `_epoch_day`. +- Create `src/fxhnt/application/funding_backtest_runner.py` — walk-forward runner + `evaluate_funding_constructions`. +- Modify `src/fxhnt/config.py` — add `crypto_pit_dir`. +- Modify `src/fxhnt/cli.py` — add `backtest-funding`. +- Tests: `tests/unit/test_cross_sectional_funding.py`, `tests/integration/test_crypto_pit_panel.py`, `tests/integration/test_funding_backtest_runner.py`, `tests/integration/test_backtest_funding_cli.py`. + +--- + +## Task 1: crypto_pit npz panel adapter + +**Files:** Create `src/fxhnt/adapters/data/crypto_pit_panel.py`; Modify `src/fxhnt/config.py`; Test `tests/integration/test_crypto_pit_panel.py`. + +- [ ] **Step 1: config** — add to `Settings` in `src/fxhnt/config.py` (mirror the existing `backtest_warehouse_path` field): +```python + crypto_pit_dir: str = Field(default=str(_DATA_DIR / "crypto_pit")) # FXHNT_CRYPTO_PIT_DIR; survivorship-free npz panel +``` +(use the same `_DATA_DIR`/`Field` idiom already in the file.) + +- [ ] **Step 2: failing test** `tests/integration/test_crypto_pit_panel.py`: +```python +import numpy as np +from fxhnt.adapters.data.crypto_pit_panel import CryptoPitPanelSource + + +def _write(tmp_path, sym, days, close, qvol, funding): + np.savez(tmp_path / f"{sym}.npz", + day=np.array(days, dtype=np.int64), close=np.array(close, dtype=float), + qvol=np.array(qvol, dtype=float), funding=np.array(funding, dtype=float)) + + +def test_loads_panel_from_npz(tmp_path): + _write(tmp_path, "AAA", [100, 101], [1.0, 1.1], [2e6, 2e6], [0.001, 0.0012]) + _write(tmp_path, "DEAD", [100], [5.0], [1e6], [0.003]) + src = CryptoPitPanelSource(str(tmp_path)) + panel = src.panel() + assert set(panel) == {"AAA", "DEAD"} + assert panel["AAA"][100] == (1.0, 2e6, 0.001) # (close, qvol, funding) + assert panel["DEAD"][100] == (5.0, 1e6, 0.003) + assert sorted(src.all_days()) == [100, 101] # union of all symbols' days + + +def test_panel_ignores_non_npz_and_is_deterministic(tmp_path): + _write(tmp_path, "BBB", [100], [1.0], [3e6], [0.0]) + (tmp_path / "notes.txt").write_text("ignore me") + a = CryptoPitPanelSource(str(tmp_path)).panel() + b = CryptoPitPanelSource(str(tmp_path)).panel() + assert a == b and set(a) == {"BBB"} +``` + +- [ ] **Step 3: run → FAIL** `.venv/bin/python -m pytest tests/integration/test_crypto_pit_panel.py -v` + +- [ ] **Step 4: implement** `src/fxhnt/adapters/data/crypto_pit_panel.py`: +```python +"""Loader for the survivorship-free `crypto_pit` npz panel (Binance USDT-perp daily +close/quote-volume/funding incl. delisted coins). Read-only; no live network.""" +from __future__ import annotations + +import glob +import os + +import numpy as np + + +class CryptoPitPanelSource: + def __init__(self, pit_dir: str) -> None: + self._dir = pit_dir + + def panel(self) -> dict[str, dict[int, tuple[float, float, float]]]: + """{symbol: {epoch_day: (close, qvol, funding)}}. Deterministic (sorted files).""" + out: dict[str, dict[int, tuple[float, float, float]]] = {} + for path in sorted(glob.glob(os.path.join(self._dir, "*.npz"))): + sym = os.path.splitext(os.path.basename(path))[0] + with np.load(path) as z: # numeric arrays only; no allow_pickle + days = z["day"].astype(np.int64) + close = z["close"].astype(float) + qvol = z["qvol"].astype(float) + funding = z["funding"].astype(float) + out[sym] = {int(d): (float(c), float(q), float(f)) + for d, c, q, f in zip(days, close, qvol, funding)} + return out + + def all_days(self) -> list[int]: + return sorted({d for s in self.panel().values() for d in s}) +``` + +- [ ] **Step 5: run → PASS**, then `.venv/bin/python -m mypy src/fxhnt/adapters/data/crypto_pit_panel.py src/fxhnt/config.py` → no new errors. + +- [ ] **Step 6: commit** +```bash +git add src/fxhnt/adapters/data/crypto_pit_panel.py src/fxhnt/config.py tests/integration/test_crypto_pit_panel.py +git commit -m "feat(xsfunding): crypto_pit npz panel adapter (survivorship-free, read-only) + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 2: domain — PIT eligibility + cross-sectional funding score + +**Files:** Create `src/fxhnt/domain/cross_sectional_funding.py`; Test `tests/unit/test_cross_sectional_funding.py`. + +- [ ] **Step 1: failing tests** `tests/unit/test_cross_sectional_funding.py`: +```python +from fxhnt.domain.cross_sectional_funding import eligible_asof, funding_score, _epoch_day + +# panel: {symbol: {epoch_day: (close, qvol, funding)}} +def _coin(start, n, qvol, funding, close=1.0): + return {start + i: (close, qvol, funding) for i in range(n)} + + +def test_eligible_filters_by_liquidity_and_history_pit(): + panel = { + "LIQ": _coin(100, 40, 5e6, 0.001), # liquid, long history + "THIN": _coin(100, 40, 1e5, 0.001), # too illiquid + "NEW": _coin(135, 40, 5e6, 0.001), # not enough history before asof + } + elig = eligible_asof(panel, asof_day=139, min_qvol=1e6, min_history=20, vol_window=30) + assert elig == ["LIQ"] # THIN below floor, NEW lacks 20d history by 139 + + +def test_funding_score_ranks_high_funding_higher(): + # three liquid coins, different trailing funding; score = robust_z of trailing-mean funding + panel = { + "HI": _coin(100, 30, 5e6, 0.003), + "MID": _coin(100, 30, 5e6, 0.001), + "LO": _coin(100, 30, 5e6, -0.001), + } + elig = ["HI", "MID", "LO"] + sc = funding_score(panel, elig, asof_day=129, lookback_days=7) + assert sc["HI"] > sc["MID"] > sc["LO"] + + +def test_funding_score_is_pit_no_lookahead(): + # LATE: funding low early, high late. At an EARLY asof it must NOT rank above a steady HI. + panel = { + "HI": {100 + i: (1.0, 5e6, 0.002) for i in range(60)}, + "LATE": {100 + i: (1.0, 5e6, (0.0001 if i < 30 else 0.01)) for i in range(60)}, + } + early = funding_score(panel, ["HI", "LATE"], asof_day=120, lookback_days=7) # day120: LATE still in low regime + assert early["HI"] > early["LATE"] + late = funding_score(panel, ["HI", "LATE"], asof_day=159, lookback_days=7) # day159: LATE now high + assert late["LATE"] > late["HI"] + + +def test_epoch_day_roundtrip(): + assert _epoch_day("1970-01-02") == 1 +``` + +- [ ] **Step 2: run → FAIL** + +- [ ] **Step 3: implement** `src/fxhnt/domain/cross_sectional_funding.py` (part 1): +```python +"""Pure cross-sectional crypto funding-dispersion domain: PIT eligibility, the +cross-sectional funding score, construction weights, and signed delta-neutral +carry book-return. No I/O. Panel = {symbol: {epoch_day: (close, qvol, funding)}}.""" +from __future__ import annotations + +import datetime as dt + +from fxhnt.domain.strategies.equity_factor import robust_z + +Panel = dict[str, dict[int, tuple[float, float, float]]] + + +def _epoch_day(iso_date: str) -> int: + return (dt.date.fromisoformat(iso_date) - dt.date(1970, 1, 1)).days + + +def eligible_asof(panel: Panel, asof_day: int, *, min_qvol: float, + min_history: int, vol_window: int = 30) -> list[str]: + """Coins tradeable at `asof_day` (PIT): have >= `min_history` bars on/before asof, + AND trailing-`vol_window` mean qvol >= `min_qvol`. Uses ONLY data <= asof_day.""" + out: list[str] = [] + lo = asof_day - vol_window + for sym, series in panel.items(): + hist = [d for d in series if d <= asof_day] + if len(hist) < min_history: + continue + vols = [series[d][1] for d in hist if d > lo] + if vols and (sum(vols) / len(vols)) >= min_qvol: + out.append(sym) + out.sort() + return out + + +def funding_score(panel: Panel, eligible: list[str], asof_day: int, + lookback_days: int = 7) -> dict[str, float]: + """Cross-sectional carry score: robust_z (across `eligible`) of each coin's + trailing-`lookback_days` mean funding (<= asof_day). Short window keeps it + responsive (avoids stale-level bias). Returns {symbol: score} aligned to `eligible`.""" + lo = asof_day - lookback_days + tf: list[float | None] = [] + for sym in eligible: + series = panel[sym] + vals = [series[d][2] for d in series if lo < d <= asof_day] + tf.append(sum(vals) / len(vals) if vals else None) + z = robust_z(tf) + return {eligible[i]: z[i] for i in range(len(eligible))} +``` + +- [ ] **Step 4: run → PASS** `.venv/bin/python -m pytest tests/unit/test_cross_sectional_funding.py -v` + +- [ ] **Step 5: mypy** on the new module → no errors. + +- [ ] **Step 6: commit** +```bash +git add src/fxhnt/domain/cross_sectional_funding.py tests/unit/test_cross_sectional_funding.py +git commit -m "feat(xsfunding): PIT eligibility + cross-sectional funding score (robust_z of trailing funding) + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 3: domain — construction weights (3 modes) + signed carry book-return + +**Files:** Modify `src/fxhnt/domain/cross_sectional_funding.py`; Test `tests/unit/test_cross_sectional_funding.py` (append). + +- [ ] **Step 1: append failing tests:** +```python +from fxhnt.domain.cross_sectional_funding import construction_weights, book_return_xs + +# scores aligned to a symbol list +SC = {"A": 2.0, "B": 1.0, "C": 0.0, "D": -1.0, "E": -2.0} # A richest carry, E poorest + + +def test_long_tilt_longs_top_quantile_sums_to_one(): + w = construction_weights(SC, "long_tilt", quantile=0.4) # top 40% of 5 = top 2 = A,B + assert set(k for k, v in w.items() if v != 0) == {"A", "B"} + assert all(v > 0 for v in w.values() if v != 0) + assert abs(sum(w.values()) - 1.0) < 1e-9 + + +def test_market_neutral_longs_top_shorts_bottom_sums_to_zero(): + w = construction_weights(SC, "market_neutral", quantile=0.4) # long A,B ; short D,E + assert w["A"] > 0 and w["B"] > 0 and w["D"] < 0 and w["E"] < 0 + assert abs(sum(w.values())) < 1e-9 # dollar-neutral + assert abs(sum(abs(v) for v in w.values()) - 2.0) < 1e-9 # gross 2.0 + + +def test_executable_shorts_only_borrowable_names(): + # only A,B,C are "borrowable"; market_neutral would short D,E but executable can't + w = construction_weights(SC, "executable", quantile=0.4, borrowable={"A", "B", "C"}) + assert w["A"] > 0 and w["B"] > 0 # long side intact + assert w.get("D", 0.0) == 0.0 and w.get("E", 0.0) == 0.0 # un-borrowable shorts dropped + + +def test_book_return_signed_carry_minus_cost(): + # long A (collect +funding), short E (collect -funding => profit when E funding negative) + prev = {"A": 0.5, "E": -0.5} + funding_today = {"A": 0.002, "E": -0.002} + new = {"A": 0.5, "E": -0.5} # no turnover + cost = {"A": 0.001, "E": 0.001} + r = book_return_xs(prev, funding_today, new, cost) + # realized = 0.5*0.002 + (-0.5)*(-0.002) = 0.001 + 0.001 = 0.002 ; no turnover cost + assert abs(r - 0.002) < 1e-12 + + +def test_book_return_charges_turnover(): + prev = {"A": 1.0} + new = {"B": 1.0} # full switch: turnover |0-1|_A + |1-0|_B = 2.0 + cost = {"A": 0.001, "B": 0.001} + r = book_return_xs(prev, {"A": 0.0}, new, cost) + # realized 0 ; cost = (|1|*0.001 + |1|*0.001)/2 = 0.001 + assert abs(r - (-0.001)) < 1e-12 +``` + +- [ ] **Step 2: run → FAIL** + +- [ ] **Step 3: implement** (append to `src/fxhnt/domain/cross_sectional_funding.py`): +```python +def _quantile_cut(scores: dict[str, float], quantile: float) -> tuple[list[str], list[str]]: + """Return (top, bottom) symbol lists at the given quantile (by score desc).""" + ranked = sorted(scores, key=lambda s: (-scores[s], s)) + k = max(1, int(quantile * len(ranked))) + return ranked[:k], ranked[-k:] + + +def construction_weights(scores: dict[str, float], mode: str, *, quantile: float = 0.2, + borrowable: set[str] | None = None) -> dict[str, float]: + """Map cross-sectional funding scores -> delta-neutral weights. + long_tilt: equal-weight long the top quantile, sum=1. + market_neutral: long top + short bottom, dollar-neutral (sum=0, gross=2). + executable: long top (sum=1) + short bottom ONLY for names in `borrowable`.""" + top, bot = _quantile_cut(scores, quantile) + w: dict[str, float] = {} + if mode == "long_tilt": + for s in top: + w[s] = 1.0 / len(top) + elif mode == "market_neutral": + for s in top: + w[s] = 0.5 / len(top) + for s in bot: + w[s] = w.get(s, 0.0) - 0.5 / len(bot) + elif mode == "executable": + for s in top: + w[s] = 1.0 / len(top) + shorts = [s for s in bot if borrowable is None or s in borrowable] + for s in shorts: + w[s] = w.get(s, 0.0) - 1.0 / len(shorts) if shorts else w.get(s, 0.0) + else: + raise ValueError(f"unknown construction mode: {mode}") + return {s: v for s, v in w.items() if v != 0.0} + + +def book_return_xs(prev_weights: dict[str, float], funding_today: dict[str, float], + new_weights: dict[str, float], coin_cost_rt: dict[str, float]) -> float: + """Signed delta-neutral carry on the PRIOR book at today's funding, minus round-trip + cost on turnover. realized = sum(w_i * funding_i); a short (w<0) on a negative-funding + coin earns w*funding>0. cost = sum(|Δw_i| * cost_i)/2 over the union.""" + realized = sum(w * funding_today.get(s, 0.0) for s, w in prev_weights.items()) + cost = sum(abs(new_weights.get(s, 0.0) - prev_weights.get(s, 0.0)) * coin_cost_rt.get(s, 0.0) + for s in set(new_weights) | set(prev_weights)) / 2.0 + return realized - cost +``` + +- [ ] **Step 4: run → PASS** + +- [ ] **Step 5: mypy** → no errors. + +- [ ] **Step 6: commit** +```bash +git add src/fxhnt/domain/cross_sectional_funding.py tests/unit/test_cross_sectional_funding.py +git commit -m "feat(xsfunding): construction weights (long_tilt/market_neutral/executable) + signed carry book-return + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 4: walk-forward backtest runner + +**Files:** Create `src/fxhnt/application/funding_backtest_runner.py`; Test `tests/integration/test_funding_backtest_runner.py`. + +- [ ] **Step 1: failing test** (synthetic panel where high-funding coins persist → long_tilt should be net positive): +```python +import numpy as np +from fxhnt.application.funding_backtest_runner import FundingBacktestRunner + + +def _panel(): + # 4 coins, 200 days; HI/MID positive funding, LO/NEG low/negative; all liquid + panel = {} + fund = {"HI": 0.0015, "MID": 0.0008, "LO": 0.0001, "NEG": -0.0010} + for sym, f in fund.items(): + panel[sym] = {1000 + i: (1.0, 5e6, f) for i in range(200)} + return panel + + +def test_runner_produces_series_per_mode(): + runner = FundingBacktestRunner(_panel(), min_qvol=1e6, min_history=20, + lookback_days=7, quantile=0.5, cost_bps=8.0, slip_coef=0.0) + res = runner.run() + assert set(res.returns_by_mode) == {"long_tilt", "market_neutral", "executable"} + for series in res.returns_by_mode.values(): + assert len(series) > 50 and np.isfinite(series).all() + # long_tilt holds the high-funding coins -> positive cumulative carry + assert float(np.sum(res.returns_by_mode["long_tilt"])) > 0.0 + + +def test_runner_deterministic(): + r1 = FundingBacktestRunner(_panel(), min_qvol=1e6, min_history=20, lookback_days=7, + quantile=0.5, cost_bps=8.0, slip_coef=0.0).run().returns_by_mode["long_tilt"] + r2 = FundingBacktestRunner(_panel(), min_qvol=1e6, min_history=20, lookback_days=7, + quantile=0.5, cost_bps=8.0, slip_coef=0.0).run().returns_by_mode["long_tilt"] + assert np.array_equal(r1, r2) +``` + +- [ ] **Step 2: run → FAIL** + +- [ ] **Step 3: implement** `src/fxhnt/application/funding_backtest_runner.py`: +```python +"""Walk-forward cross-sectional funding-dispersion backtest over the crypto_pit panel. +Daily rebalance, PIT universe, signed delta-neutral carry minus a liquidity-scaled cost. +Produces a daily net-return series per construction mode.""" +from __future__ import annotations + +from dataclasses import dataclass + +import numpy as np + +from fxhnt.domain.cross_sectional_funding import ( + Panel, book_return_xs, construction_weights, eligible_asof, funding_score, +) + +_MODES = ("long_tilt", "market_neutral", "executable") + + +@dataclass(frozen=True) +class FundingRunResult: + returns_by_mode: dict[str, np.ndarray] + n_names_avg: float + trading_days: int + + +class FundingBacktestRunner: + def __init__(self, panel: Panel, *, min_qvol: float = 1e6, min_history: int = 30, + lookback_days: int = 7, quantile: float = 0.2, cost_bps: float = 8.0, + slip_coef: float = 0.0005, slip_cap_bps: float = 50.0, + liq_ref: float = 1e6, borrowable_qvol: float = 2e7) -> None: + self._p = panel + self._minq, self._minh, self._lb, self._q = min_qvol, min_history, lookback_days, quantile + self._cost = cost_bps / 1e4 + self._slip_coef, self._slip_cap = slip_coef, slip_cap_bps / 1e4 + self._liq_ref, self._borrow_q = liq_ref, borrowable_qvol + + def _coin_cost(self, qvol: float) -> float: + """Round-trip cost: taker fee both legs + liquidity-scaled slippage (thin=more).""" + slip = min(self._slip_cap, self._slip_coef * (self._liq_ref / qvol)) if qvol > 0 else self._slip_cap + return self._cost + slip + + def run(self) -> FundingRunResult: + all_days = sorted({d for s in self._p.values() for d in s}) + series: dict[str, list[float]] = {m: [] for m in _MODES} + prev_w: dict[str, dict[str, float]] = {m: {} for m in _MODES} + names_seen = 0 + n_rebals = 0 + for i in range(len(all_days) - 1): + d, nxt = all_days[i], all_days[i + 1] + elig = eligible_asof(self._p, d, min_qvol=self._minq, min_history=self._minh) + if len(elig) < 4: # need a cross-section to rank + continue + scores = funding_score(self._p, elig, d, lookback_days=self._lb) + borrowable = {s for s in elig if self._p[s].get(d, (0, 0, 0))[1] >= self._borrow_q} + funding_next = {s: self._p[s][nxt][2] for s in elig if nxt in self._p[s]} + cost = {s: self._coin_cost(self._p[s][d][1]) for s in elig} + names_seen += len(elig) + n_rebals += 1 + for m in _MODES: + w = construction_weights(scores, m, quantile=self._q, borrowable=borrowable) + # carry the NEW book earns over [d, nxt] at next-day funding, minus rebalance cost vs prior book + series[m].append(_book(w, prev_w[m], funding_next, cost)) + prev_w[m] = w + return FundingRunResult( + returns_by_mode={m: np.asarray(series[m], dtype=float) for m in _MODES}, + n_names_avg=(names_seen / n_rebals) if n_rebals else 0.0, + trading_days=len(all_days), + ) + + +def _book(w: dict[str, float], prev_w: dict[str, float], + funding_next: dict[str, float], cost: dict[str, float]) -> float: + """Carry the NEW book `w` earns over the next day (sum w_i * funding_next_i) minus the + round-trip rebalance cost charged on turnover between the prior book and the new one.""" + realized = sum(wt * funding_next.get(s, 0.0) for s, wt in w.items()) + 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 +``` +(`book_return_xs` stays as the domain primitive exercised by the Task-3 unit tests; the runner uses this direct, clearer `_book`.) + +- [ ] **Step 4: run → PASS.** If `long_tilt` isn't positive, debug by printing the realized series; do NOT weaken the determinism test. Adjust the synthetic fixture only if the quantile/360-day math genuinely needs it (report why). + +- [ ] **Step 5: mypy** → no errors. + +- [ ] **Step 6: commit** +```bash +git add src/fxhnt/application/funding_backtest_runner.py tests/integration/test_funding_backtest_runner.py +git commit -m "feat(xsfunding): walk-forward runner — PIT universe + per-mode daily carry series + liquidity-scaled cost + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 5: gauntlet verdict matrix (modes × liquidity floors) + +**Files:** Modify `src/fxhnt/application/funding_backtest_runner.py` (add `evaluate_funding_matrix`); Test `tests/integration/test_funding_backtest_runner.py` (append). + +- [ ] **Step 1: append failing test:** +```python +from fxhnt.application.funding_backtest_runner import evaluate_funding_matrix + + +def test_evaluate_matrix_runs_all_cells(tmp_path): + panel = _panel() + 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) + # 3 modes x 2 floors = 6 cells + assert len(report["cells"]) == 6 + for key, cell in report["cells"].items(): + 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) # JSON-serializable (no NaN/np types) +``` + +- [ ] **Step 2: run → FAIL** + +- [ ] **Step 3: implement** (append to `src/fxhnt/application/funding_backtest_runner.py`): +```python +from typing import Any + +from fxhnt.domain.backtest import compute_stats +from fxhnt.domain.gauntlet.core import evaluate, per_period_sharpe + +_PERIODS_PER_YEAR = 365 # crypto trades daily, all year + + +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) + # cross-cell sr_variance (Bailey/Lopez de Prado), matching discovery.py convention + 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": 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": stats.model_dump(), "verdict": v.model_dump()} + return out +``` + +- [ ] **Step 4: run → PASS.** Verify `per_period_sharpe` is the real gauntlet helper name (check `domain/gauntlet/core.py` + how `application/discovery.py` computes `sr_variance`); if it differs, use the real one. Confirm `model_dump()` is JSON-native (no numpy scalars); if `compute_stats`/`evaluate` leak `np.float64`, coerce with `float(...)`. + +- [ ] **Step 5: mypy** → no errors; full `.venv/bin/python -m pytest -q` → all PASS. + +- [ ] **Step 6: commit** +```bash +git add src/fxhnt/application/funding_backtest_runner.py tests/integration/test_funding_backtest_runner.py +git commit -m "feat(xsfunding): gauntlet verdict matrix (modes x floors, crypto 365d, cross-cell DSR n_trials) + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 6: CLI `backtest-funding` + run the real verdict + +**Files:** Modify `src/fxhnt/cli.py`; Test `tests/integration/test_backtest_funding_cli.py`. + +- [ ] **Step 1: read** `src/fxhnt/cli.py` `backtest-equity` for the house idiom (`get_settings()`, function-local imports, `--out` json write, echo loop). + +- [ ] **Step 2: failing CLI test** (mirror `test_backtest_equity_cli.py`): build a synthetic crypto_pit dir of npz (reuse the `_panel()` shape, write each coin to `.npz`), set `FXHNT_CRYPTO_PIT_DIR` to it (monkeypatch + `cli.get_settings.cache_clear()`), invoke `backtest-funding --floors 1e6 --out `, assert exit 0 and the report JSON has cells for the 3 modes. +```python +import json +import numpy as np +from typer.testing import CliRunner +from fxhnt.cli import app + + +def _write_pit(d): + fund = {"HI": 0.0015, "MID": 0.0008, "LO": 0.0001, "NEG": -0.0010} + for sym, f in fund.items(): + days = np.array([1000 + i for i in range(200)], dtype=np.int64) + np.savez(d / f"{sym}.npz", day=days, close=np.ones(200), qvol=np.full(200, 5e6), + funding=np.full(200, f)) + + +def test_cli_backtest_funding_writes_report(tmp_path, monkeypatch): + pit = tmp_path / "pit"; pit.mkdir(); _write_pit(pit) + out = tmp_path / "funding_report.json" + monkeypatch.setenv("FXHNT_CRYPTO_PIT_DIR", str(pit)) + from fxhnt import cli as climod + climod.get_settings.cache_clear() + res = CliRunner().invoke(app, ["backtest-funding", "--floors", "1e6", "--quantile", "0.5", + "--cost-bps", "8", "--out", str(out)]) + assert res.exit_code == 0, res.output + report = json.loads(out.read_text()) + modes = {c["mode"] for c in report["cells"].values()} + assert modes == {"long_tilt", "market_neutral", "executable"} +``` + +- [ ] **Step 3: run → FAIL** + +- [ ] **Step 4: implement** the command in `src/fxhnt/cli.py`: +```python +@app.command("backtest-funding") +def backtest_funding( + floors: list[float] = typer.Option([1e6, 5e6, 2e7], "--floors", help="Liquidity floors ($/day) to sweep."), + lookback_days: int = typer.Option(7, "--lookback-days", help="Trailing window for the funding score."), + quantile: float = typer.Option(0.2, "--quantile", help="Top/bottom quantile for construction."), + cost_bps: float = typer.Option(8.0, "--cost-bps", help="Taker round-trip cost (both legs), bps."), + slip_coef: float = typer.Option(0.0005, "--slip-coef", help="Liquidity-scaled slippage coefficient."), + out: str = typer.Option("/backtest-data/funding_backtest_report.json", "--out", help="Report JSON path."), +) -> None: + """Cross-sectional crypto funding-dispersion backtest: verdict matrix (modes x liquidity floors) + over the survivorship-free crypto_pit panel.""" + import json + from fxhnt.adapters.data.crypto_pit_panel import CryptoPitPanelSource + from fxhnt.application.funding_backtest_runner import evaluate_funding_matrix + + s = get_settings() + panel = CryptoPitPanelSource(s.crypto_pit_dir).panel() + report = evaluate_funding_matrix(panel, floors=floors, lookback_days=lookback_days, quantile=quantile, + cost_bps=cost_bps, slip_coef=slip_coef, + oos_fraction=s.gauntlet.oos_fraction, dsr_min=s.gauntlet.dsr_min, + oos_min_sharpe=s.gauntlet.oos_min_sharpe, + max_is_oos_decay=s.gauntlet.max_is_oos_decay) + with open(out, "w") as fh: + json.dump(report, fh, indent=2) + for key, c in sorted(report["cells"].items()): + v, st = c["verdict"], c["stats"] + typer.echo(f"backtest-funding {key}: passed={v['passed']} dsr={v['dsr']:.3f} " + f"sharpe={st['sharpe']:.2f} cagr={st['cagr']*100:.1f}% maxDD={st['max_drawdown']*100:.1f}% -> {out}") +``` +(Match the real `get_settings()` idiom + function-local imports from `backtest-equity`; there's no store to close here.) + +- [ ] **Step 5: run → PASS**, full `.venv/bin/python -m pytest -q` → all PASS, mypy on cli.py → no NEW errors. + +- [ ] **Step 6: commit** +```bash +git add src/fxhnt/cli.py tests/integration/test_backtest_funding_cli.py +git commit -m "feat(xsfunding): backtest-funding CLI — verdict matrix over crypto_pit panel + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +- [ ] **Step 7: RUN THE REAL VERDICT** (controller does this, not a subagent): point at the on-disk panel and run the matrix: +```bash +FXHNT_CRYPTO_PIT_DIR=/home/jgrusewski/Work/foxhunt/data/surfer/crypto_pit \ + .venv/bin/fxhnt backtest-funding --floors 1e6 --floors 5e6 --floors 2e7 \ + --quantile 0.2 --cost-bps 8 --out /tmp/funding_report.json +``` +Report the verdict matrix; apply the pre-registered kill criteria (DSR<0.5 / edge only sub-$1M-ADV / corr>0.6 to existing funding+momentum). + +--- + +## Known caveats (flag, do not fix here) +- Delisted-ticker roster = the 35 famous blow-ups in the panel (survivorship-free for those, not an exhaustive census). Documented; future `data.binance.vision` hardening. +- Daily-summed funding only (no 8h). Daily rebalance. +- Cost model is a liquidity-scaled approximation, not order-book-exact — deliberately conservative. + +## Final review (after Task 6) +Final code review over the whole diff; full suite + mypy. Then interpret the verdict and decide on the paper-forward follow-on. Do NOT push.