docs(xsfunding): spot-perp basis-modeling implementation plan

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-06-19 00:12:47 +02:00
parent 2347714cae
commit 178ac665c4

View File

@@ -0,0 +1,360 @@
# Funding Backtest — Spot-Perp Basis Modeling — Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: superpowers:subagent-driven-development. Steps use `- [ ]`.
**Goal:** Extend the cross-sectional funding backtest to model the spot-perp BASIS P&L (the dominant missing risk), turning the funding-only Sharpe into the real net-of-basis number.
**Architecture:** Add a SPOT price leg (fetched from Binance spot klines) alongside the panel's existing PERP `close`, guarded against the delisted-ticker-reuse trap (clip spot to each coin's perp-alive window + reject |basis|>30% decoupling days). The runner's per-coin return becomes the true delta-neutral cash-and-carry `(spot_ret perp_ret + funding)`, with funding-only fallback where spot is unavailable. Re-run the verdict matrix.
**Tech Stack:** Python 3.12, numpy, Binance spot klines (`api/v3/klines`), the existing crypto_pit panel + funding runner/gauntlet, pytest, strict mypy.
**Spike (done):** `close`=PERP; spot is the missing leg; spot obtainable incl. delisted BUT ticker-reuse trap (spot LUNAUSDT→LUNA2.0) MUST be guarded; alt Δbasis vol is 1.65.5× funding vol → basis materially changes the Sharpe.
**House rules:** TDD, pure functions in domain, NO live network in tests (fake clients / synthetic panels), strict mypy (no NEW errors), commit per task (do NOT push), trailer `Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>`. Crypto 365/yr.
**Backward-compat:** when no spot panel is supplied, the runner must behave EXACTLY as today (funding-only) — all existing tests stay green.
---
## Reuse contract
```python
# domain/cross_sectional_funding.py
Panel = dict[str, dict[int, tuple[float, float, float]]] # {sym:{day:(perp_close, qvol, funding)}}
# application/funding_backtest_runner.py
class FundingBacktestRunner: __init__(panel, *, min_qvol, min_history, lookback_days, quantile, cost_bps, slip_coef, ...)
.run() -> FundingRunResult(returns_by_mode, n_names_avg, panel_days, n_rebalances)
def evaluate_funding_matrix(panel, *, floors, ...) -> dict
# adapters/data/crypto_pit_panel.py : CryptoPitPanelSource(pit_dir).panel()
```
New type: `SpotPanel = dict[str, dict[int, float]]` ({sym:{epoch_day:spot_close}}).
---
## File Structure
- Create `src/fxhnt/adapters/data/binance_spot_history.py``BinanceSpotHistory` (fetch spot daily close, paginated).
- Create `src/fxhnt/application/basis_panel_builder.py` — orchestrate: perp panel + spot fetch → cleaned `SpotPanel`; pure guard `clean_spot_leg`.
- Modify `src/fxhnt/domain/cross_sectional_funding.py` — add pure `clean_spot_leg` + `SpotPanel` alias.
- Modify `src/fxhnt/application/funding_backtest_runner.py``_book` generalized to a per-coin return dict; runner accepts `spot: SpotPanel | None`; `evaluate_funding_matrix` accepts/threads spot.
- Modify `src/fxhnt/adapters/data/crypto_pit_panel.py` — add `spot_panel(spot_dir)` loader (reads the spot npz sidecar).
- Modify `src/fxhnt/cli.py``build-basis-panel` command + `backtest-funding --spot-dir`.
- Tests under `tests/unit/` + `tests/integration/`.
---
## Task 1: spot-history adapter + the ticker-reuse guard (pure)
**Files:** Create `src/fxhnt/adapters/data/binance_spot_history.py`; Modify `src/fxhnt/domain/cross_sectional_funding.py`; Tests `tests/unit/test_cross_sectional_funding.py` (append), `tests/integration/test_binance_spot_history.py`.
- [ ] **Step 1: pure guard test** (append to `tests/unit/test_cross_sectional_funding.py`):
```python
from fxhnt.domain.cross_sectional_funding import clean_spot_leg
def test_clean_spot_clips_to_perp_window_and_rejects_relist():
perp_days = [100, 101, 102, 103] # perp alive days 100-103
perp_close = {100: 1.00, 101: 1.00, 102: 1.00, 103: 1.00}
# spot has a pre-window day (99), in-window clean days, a relist-jump day (102 → basis 50x), and a post-window day (200)
spot_raw = {99: 1.0, 100: 1.001, 101: 0.999, 102: 50.0, 103: 1.000, 200: 8.87}
clean = clean_spot_leg(perp_days, perp_close, spot_raw, max_basis=0.30)
assert set(clean) == {100, 101, 103} # 99/200 out of window; 102 rejected (|basis|>30%)
assert clean[100] == 1.001
def test_clean_spot_empty_when_no_overlap():
assert clean_spot_leg([100, 101], {100: 1.0, 101: 1.0}, {200: 5.0}, max_basis=0.30) == {}
```
- [ ] **Step 2: run → FAIL.**
- [ ] **Step 3: implement** `clean_spot_leg` (append to `src/fxhnt/domain/cross_sectional_funding.py`):
```python
SpotPanel = dict[str, dict[int, float]]
def clean_spot_leg(perp_days: list[int], perp_close: dict[int, float],
spot_raw: dict[int, float], *, max_basis: float = 0.30) -> dict[int, float]:
"""Cleaned spot leg for one coin: keep spot ONLY on days within the perp's alive window
[min(perp_days), max(perp_days)] AND where |basis|=|(perp-spot)/spot| <= max_basis.
Drops pre-listing/post-delisting spot bars and ticker-reuse/relist decoupling artifacts
(e.g. spot LUNAUSDT becoming LUNA 2.0). Days dropped here fall back to funding-only downstream."""
if not perp_days:
return {}
lo, hi = min(perp_days), max(perp_days)
out: dict[int, float] = {}
for d, sp in spot_raw.items():
if d < lo or d > hi or sp <= 0:
continue
pc = perp_close.get(d)
if pc is None or pc <= 0:
continue
if abs((pc - sp) / sp) <= max_basis:
out[d] = sp
return out
```
- [ ] **Step 4: spot adapter test** `tests/integration/test_binance_spot_history.py` (fake HTTP, no live net) — mirror how `binance_funding.py` is tested if a pattern exists; otherwise inject a fake fetch:
```python
from fxhnt.adapters.data.binance_spot_history import BinanceSpotHistory
def test_spot_history_parses_klines(monkeypatch):
# kline row: [openTime, open, high, low, close, vol, closeTime, quoteVol, ...]; close=index 4, openTime ms
rows = [[100 * 86400000, "1", "1", "1", "1.5", "0", 0, "0"],
[101 * 86400000, "1", "1", "1", "1.7", "0", 0, "0"]]
h = BinanceSpotHistory()
monkeypatch.setattr(h, "_get", lambda url: rows) # one page then empty
calls = {"n": 0}
def fake_get(url):
calls["n"] += 1
return rows if calls["n"] == 1 else []
monkeypatch.setattr(h, "_get", fake_get)
out = h.daily_close("AAAUSDT")
assert out == {100: 1.5, 101: 1.7} # epoch-DAY keys, close values
```
- [ ] **Step 5: implement** `src/fxhnt/adapters/data/binance_spot_history.py`:
```python
"""Binance SPOT daily-close history (the long leg of the cash-and-carry), paginated.
Returns {epoch_day: close}. Live network in production; tests inject a fake `_get`."""
from __future__ import annotations
import json
import time
import urllib.request
from typing import Any
_SPOT = "https://api.binance.com"
class BinanceSpotHistory:
def _get(self, url: str, tries: int = 4) -> Any:
last: Exception | None = None
for i in range(tries):
try:
with urllib.request.urlopen(url, timeout=30) as r:
return json.loads(r.read())
except Exception as e: # network/HTTP — retry with backoff
last = e
time.sleep(0.5 * (i + 1))
raise RuntimeError(f"spot fetch failed: {url}: {last}")
def daily_close(self, symbol: str, start_ms: int = 0) -> dict[int, float]:
"""Full daily spot-close history {epoch_day: close}, forward-paginated (limit 1000)."""
out: dict[int, float] = {}
st = start_ms
while True:
url = (f"{_SPOT}/api/v3/klines?symbol={symbol}&interval=1d"
f"&startTime={st}&limit=1000")
rows = self._get(url)
if not rows:
break
for r in rows:
out[int(r[0]) // 86_400_000] = float(r[4])
nxt = int(rows[-1][0]) + 86_400_000
if nxt <= st or len(rows) < 1000:
st = nxt
if len(rows) < 1000:
break
st = nxt
return out
```
(NOTE: keep the pagination simple + correct — advance `st` past the last row's openTime; stop when a page returns <1000 rows or empty. The test's fake returns one full-looking page then empty; ensure the loop terminates. Adjust the loop so it terminates cleanly for both the fake and real Binance; the key invariant: every row's `openTime_ms // 86_400_000` is the epoch-day key and `r[4]` is the close.)
- [ ] **Step 6: run both test files → PASS;** mypy on the 2 files → no errors; full suite → PASS.
- [ ] **Step 7: commit**
```bash
git add src/fxhnt/adapters/data/binance_spot_history.py src/fxhnt/domain/cross_sectional_funding.py tests/unit/test_cross_sectional_funding.py tests/integration/test_binance_spot_history.py
git commit -m "feat(xsfunding): spot-history adapter + ticker-reuse guard clean_spot_leg (basis prep)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
```
---
## Task 2: basis-panel builder + CLI
**Files:** Create `src/fxhnt/application/basis_panel_builder.py`; Modify `src/fxhnt/adapters/data/crypto_pit_panel.py` (add `spot_panel` loader); Modify `src/fxhnt/cli.py` (`build-basis-panel`); Tests `tests/integration/test_basis_panel_builder.py`.
- [ ] **Step 1: failing test** `tests/integration/test_basis_panel_builder.py`:
```python
import numpy as np
from fxhnt.application.basis_panel_builder import build_spot_panel
class _FakeSpot:
def __init__(self, data): self._d = data
def daily_close(self, symbol, start_ms=0): return dict(self._d.get(symbol, {}))
def test_build_spot_panel_applies_guard():
# perp panel: AAA alive days 100-103; spot has an in-window relist jump on 102 + a post-window 200
panel = {"AAA": {100: (1.0, 5e6, 0.001), 101: (1.0, 5e6, 0.001),
102: (1.0, 5e6, 0.001), 103: (1.0, 5e6, 0.001)}}
spot = _FakeSpot({"AAA": {100: 1.001, 101: 0.999, 102: 50.0, 103: 1.0, 200: 8.87}})
sp = build_spot_panel(panel, spot, max_basis=0.30)
assert set(sp["AAA"]) == {100, 101, 103} # 102 rejected, 200 out of window
```
- [ ] **Step 2: run → FAIL.**
- [ ] **Step 3: implement** `src/fxhnt/application/basis_panel_builder.py`:
```python
"""Build the cleaned SPOT leg for the funding backtest: fetch spot history per panel symbol,
apply the perp-window + |basis| guard (clean_spot_leg), return a SpotPanel."""
from __future__ import annotations
from fxhnt.domain.cross_sectional_funding import Panel, SpotPanel, clean_spot_leg
class _SpotClient: # structural typing hint
def daily_close(self, symbol: str, start_ms: int = 0) -> dict[int, float]: ...
def build_spot_panel(panel: Panel, spot_client: _SpotClient, *, max_basis: float = 0.30) -> SpotPanel:
out: SpotPanel = {}
for sym, series in panel.items():
perp_days = sorted(series)
perp_close = {d: series[d][0] for d in perp_days}
try:
raw = spot_client.daily_close(sym)
except Exception:
continue # spot unavailable for this coin -> funding-only downstream
cleaned = clean_spot_leg(perp_days, perp_close, raw, max_basis=max_basis)
if cleaned:
out[sym] = cleaned
return out
```
- [ ] **Step 4: spot-panel npz persistence** — add to `CryptoPitPanelSource` (or a sibling) a way to WRITE the spot panel to npz + a `spot_panel(spot_dir)` loader, so the CLI can persist + the backtest can reload without re-fetching. Add to `src/fxhnt/adapters/data/crypto_pit_panel.py`:
```python
@staticmethod
def write_spot_panel(spot: dict[str, dict[int, float]], spot_dir: str) -> None:
import os
os.makedirs(spot_dir, exist_ok=True)
for sym, series in spot.items():
days = sorted(series)
np.savez(os.path.join(spot_dir, f"{sym}.npz"),
day=np.array(days, dtype=np.int64),
spot=np.array([series[d] for d in days], dtype=float))
@staticmethod
def load_spot_panel(spot_dir: str) -> dict[str, dict[int, float]]:
import glob, os
out: dict[str, dict[int, float]] = {}
for path in sorted(glob.glob(os.path.join(spot_dir, "*.npz"))):
sym = os.path.splitext(os.path.basename(path))[0]
with np.load(path) as z:
out[sym] = {int(d): float(s) for d, s in zip(z["day"].astype(np.int64), z["spot"].astype(float))}
return out
```
Add a round-trip test (write then load == original).
- [ ] **Step 5: CLI `build-basis-panel`** in `src/fxhnt/cli.py` (function-local imports, `get_settings()` idiom):
```python
@app.command("build-basis-panel")
def build_basis_panel(
spot_dir: str = typer.Option(..., "--spot-dir", help="Output dir for the cleaned spot-leg npz panel."),
max_basis: float = typer.Option(0.30, "--max-basis", help="Reject |basis| above this (decoupling/relist guard)."),
) -> None:
"""Fetch + clean the SPOT leg (perp-window + |basis| guard) for every coin in the crypto_pit panel,
writing a spot npz panel for the basis-aware funding backtest."""
from fxhnt.adapters.data.crypto_pit_panel import CryptoPitPanelSource
from fxhnt.adapters.data.binance_spot_history import BinanceSpotHistory
from fxhnt.application.basis_panel_builder import build_spot_panel
s = get_settings()
panel = CryptoPitPanelSource(s.crypto_pit_dir).panel()
spot = build_spot_panel(panel, BinanceSpotHistory(), max_basis=max_basis)
CryptoPitPanelSource.write_spot_panel(spot, spot_dir)
typer.echo(f"build-basis-panel: wrote spot leg for {len(spot)}/{len(panel)} coins -> {spot_dir}")
```
- [ ] **Step 6: run tests → PASS;** mypy → no new errors; full suite → PASS.
- [ ] **Step 7: commit**
```bash
git add src/fxhnt/application/basis_panel_builder.py src/fxhnt/adapters/data/crypto_pit_panel.py src/fxhnt/cli.py tests/integration/test_basis_panel_builder.py
git commit -m "feat(xsfunding): basis-panel builder + build-basis-panel CLI + spot-panel npz persistence
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
```
---
## Task 3: runner — true delta-neutral return (spot+perp+funding) with funding-only fallback
**Files:** Modify `src/fxhnt/application/funding_backtest_runner.py`; Modify `src/fxhnt/cli.py` (`backtest-funding --spot-dir`); Test `tests/integration/test_funding_backtest_runner.py` (append).
- [ ] **Step 1: failing test:**
```python
def test_runner_basis_changes_return_vs_funding_only():
# one coin held; spot moves AGAINST the perp (basis widens) -> net return < funding-only
panel = {}
fund = {"HI": 0.002, "B": 0.001, "C": 0.0005, "D": -0.001}
for sym, f in fund.items():
panel[sym] = {1000 + i: (100.0, 5e6, f) for i in range(60)} # perp flat at 100
# spot for HI declines 100 -> 99 over the window (long-spot leg loses -> basis P&L negative)
spot = {"HI": {1000 + i: 100.0 - 0.02 * i for i in range(60)}}
from fxhnt.application.funding_backtest_runner import FundingBacktestRunner
fo = FundingBacktestRunner(panel, min_qvol=1e6, min_history=20, lookback_days=7, quantile=0.5,
cost_bps=0.0, slip_coef=0.0).run().returns_by_mode["long_tilt"]
bm = FundingBacktestRunner(panel, spot=spot, min_qvol=1e6, min_history=20, lookback_days=7,
quantile=0.5, cost_bps=0.0, slip_coef=0.0).run().returns_by_mode["long_tilt"]
import numpy as np
assert float(np.sum(bm)) < float(np.sum(fo)) # basis loss on the long-spot leg reduces return
# coins without spot fall back to funding-only (no crash)
assert np.isfinite(bm).all()
```
- [ ] **Step 2: run → FAIL.**
- [ ] **Step 3: implement** — generalize `_book` to take a per-coin return dict; runner accepts `spot`:
- Change `_book(w, prev_w, unit_ret, cost)``realized = Σ w·unit_ret[s]`, turnover cost unchanged (rename `funding_next` param to `unit_ret`).
- `FundingBacktestRunner.__init__` gains `spot: SpotPanel | None = None` (store `self._spot`).
- In `run()`, per rebalance, build `unit_ret` per eligible coin:
```python
unit_ret: dict[str, float] = {}
for s in elig:
f = self._p[s][nxt][2] if nxt in self._p[s] else 0.0
if self._spot is not None and s in self._spot and d in self._spot[s] and nxt in self._spot[s] \
and d in self._p[s] and nxt in self._p[s] and self._p[s][d][0] > 0 and self._spot[s][d] > 0:
spot_ret = self._spot[s][nxt] / self._spot[s][d] - 1.0
perp_ret = self._p[s][nxt][0] / self._p[s][d][0] - 1.0
unit_ret[s] = (spot_ret - perp_ret) + f # delta-neutral cash-and-carry
else:
unit_ret[s] = f # funding-only fallback
```
then `series[m].append(_book(w, prev_w[m], unit_ret, cost))`.
- `evaluate_funding_matrix(panel, *, spot=None, floors, ...)` threads `spot` into each `FundingBacktestRunner(..., spot=spot)`. Add a `"basis_modeled": spot is not None` flag to the report dict + update the `funding_only` caveat text to say basis IS modeled when spot is provided.
- Backward-compat: `spot=None` → `unit_ret = funding-only` → identical to current behavior (existing tests pass).
- [ ] **Step 4: run** the new test + the full runner file → PASS. Existing funding-only tests MUST stay green (spot defaults None). mypy → no errors. Full suite → PASS.
- [ ] **Step 5: CLI wiring** — `backtest-funding` gains `--spot-dir` (optional); when given, load the spot panel and pass `spot=` to `evaluate_funding_matrix`; echo whether basis is modeled. Update the caveat echo to reflect basis-modeled vs funding-only.
- [ ] **Step 6: commit**
```bash
git add src/fxhnt/application/funding_backtest_runner.py src/fxhnt/cli.py tests/integration/test_funding_backtest_runner.py
git commit -m "feat(xsfunding): model spot-perp basis P&L in runner (true delta-neutral return; funding-only fallback)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
```
---
## Task 4: build the real spot panel + re-run the net-of-basis verdict (controller)
- [ ] Build the spot panel against the real universe (network, ~180 coins):
```bash
FXHNT_CRYPTO_PIT_DIR=/home/jgrusewski/Work/foxhunt/data/surfer/crypto_pit \
.venv/bin/fxhnt build-basis-panel --spot-dir /tmp/crypto_pit_spot --max-basis 0.30
```
- [ ] Re-run the matrix WITH basis:
```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 --spot-dir /tmp/crypto_pit_spot --out /tmp/funding_basis_report.json
```
- [ ] Report the net-of-basis verdict vs the funding-only number; apply the pre-registered kill criteria (DSR<0.5 / edge only sub-$1M / corr>0.6 to existing books). Honest verdict.
## Final review (after Task 3)
Final code review over the whole basis diff; full suite + mypy. Then interpret the net-of-basis verdict and decide next (paper-forward / cockpit wiring / prune).