feat(xsfunding): live-booking forward strategy (delta-neutral funding+basis, executable)

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

src/fxhnt/application/xsfunding_strategy.py

@@ -0,0 +1,60 @@
+"""Live-booking cross-sectional funding-dispersion paper track (executable construction).
+Books the REAL delta-neutral return (funding + spot-perp basis) forward each day, so the paper
+record experiences the basis tail the backtest had to guard out. Mirrors the equity-factor
+live tracks' ForwardStrategy contract: advance(last_date, extra) -> (rows, extra)."""
+from __future__ import annotations
+
+import datetime as dt
+from typing import Any, Callable
+
+from fxhnt.domain.cross_sectional_funding import construction_weights
+from fxhnt.domain.strategies.equity_factor import robust_z
+
+
+def _today() -> str:
+    return dt.date.today().isoformat()
+
+
+class XsFundingForward:
+    def __init__(self, live: Any, *, quantile: float = 0.2, cost_bps: float = 8.0,
+                 borrowable_qvol: float = 2e7, clock: Callable[[], str] = _today) -> None:
+        self._live = live
+        self._q = quantile
+        self._cost = cost_bps / 1e4
+        self._bq = borrowable_qvol
+        self._clock = clock
+
+    def advance(self, last_date: str | None, extra: dict[str, Any]) -> tuple[list[tuple[str, float]], dict[str, Any]]:
+        today = self._clock()
+        prev_pos: dict[str, float] = extra.get("positions", {})
+        prev_perp: dict[str, float] = extra.get("perp", {})
+        prev_spot: dict[str, float] = extra.get("spot", {})
+        snap = self._live.snapshot(prev_pos)          # {sym: (tf, today_funding, perp, spot, qvol)}
+        syms = sorted(snap)
+
+        # 1) new executable weights from the trailing-funding cross-section
+        tf = [snap[s][0] for s in syms]
+        scores = {syms[i]: z for i, z in enumerate(robust_z(tf))}
+        borrowable = {s for s in syms if snap[s][4] >= self._bq}
+        new_w = construction_weights(scores, "executable", quantile=self._q, borrowable=borrowable)
+
+        # 2) book the PRIOR book's realized delta-neutral return (funding + basis) minus rebalance cost
+        rows: list[tuple[str, float]] = []
+        if prev_pos:
+            realized = 0.0
+            for s, w in prev_pos.items():
+                if s in snap and s in prev_perp and s in prev_spot and prev_perp[s] > 0 and prev_spot[s] > 0:
+                    _, tf_today, perp, spot, _ = snap[s]
+                    spot_ret = spot / prev_spot[s] - 1.0
+                    perp_ret = perp / prev_perp[s] - 1.0
+                    realized += w * ((spot_ret - perp_ret) + tf_today)
+            turnover = sum(abs(new_w.get(s, 0.0) - prev_pos.get(s, 0.0)) for s in set(new_w) | set(prev_pos))
+            realized -= turnover * self._cost / 2.0
+            rows.append((today, realized))
+
+        # 3) carry the new book forward
+        extra = {"positions": new_w,
+                 "perp": {s: snap[s][2] for s in new_w},
+                 "spot": {s: snap[s][3] for s in new_w},
+                 "last_date": today}
+        return rows, extra

tests/integration/test_xsfunding_strategy.py

@@ -0,0 +1,46 @@
+from fxhnt.application.xsfunding_strategy import XsFundingForward
+
+
+class _FakeLive:
+    def __init__(self, seq): self._seq = seq; self._i = -1
+    def snapshot(self, prev):
+        self._i += 1
+        return self._seq[self._i]
+
+
+def _snap(funding):   # perp=spot=100 (no basis move), trailing=today=funding
+    return {s: (f, f, 100.0, 100.0, 5e6) for s, f in funding.items()}
+
+
+def test_inception_books_nothing():
+    live = _FakeLive([_snap({"A": 0.002, "B": 0.001, "C": 0.0, "D": -0.001, "E": -0.002})])
+    strat = XsFundingForward(live, quantile=0.4, clock=lambda: "2026-06-19")
+    rows, extra = strat.advance(None, {})
+    assert rows == []                       # no prior book to book at inception
+    assert extra["positions"]               # executable weights set (long top-carry)
+    assert any(w > 0 for w in extra["positions"].values())
+
+
+def test_second_day_books_delta_neutral_return():
+    day1 = _snap({"A": 0.002, "B": 0.001, "C": 0.0, "D": -0.001, "E": -0.002})
+    day2 = {"A": (0.002, 0.002, 99.0, 100.0, 5e6),   # A perp 100->99 (short-perp gains) + funding
+            "B": (0.001, 0.001, 100.0, 100.0, 5e6), "C": (0.0, 0.0, 100.0, 100.0, 5e6),
+            "D": (-0.001, -0.001, 100.0, 100.0, 5e6), "E": (-0.002, -0.002, 100.0, 100.0, 5e6)}
+    live = _FakeLive([day1, day2])
+    strat = XsFundingForward(live, quantile=0.4, cost_bps=0.0, clock=lambda: "2026-06-20")
+    _, extra = strat.advance(None, {})       # inception on day1
+    rows, extra = strat.advance("2026-06-19", extra)
+    assert len(rows) == 1 and rows[0][0] == "2026-06-20"
+    assert rows[0][1] > 0.0                   # A long: short-perp gain (1%) + funding
+
+
+def test_short_leg_profits_when_negative_funding_coin_craters():
+    # market_neutral-ish: E (most negative funding) is shorted (long perp + short spot); if E's perp rises, short loses,
+    # but here test the funding sign: E short-carry earns -w*funding>0 when funding negative. Keep prices flat.
+    day1 = _snap({"A": 0.003, "B": 0.002, "C": 0.0, "D": -0.002, "E": -0.003})
+    day2 = _snap({"A": 0.003, "B": 0.002, "C": 0.0, "D": -0.002, "E": -0.003})
+    live = _FakeLive([day1, day2])
+    strat = XsFundingForward(live, quantile=0.4, cost_bps=0.0, clock=lambda: "2026-06-20")
+    _, extra = strat.advance(None, {})
+    rows, _ = strat.advance("2026-06-19", extra)
+    assert rows[0][1] > 0.0                   # long A (+0.003 funding) and short E (-w*-0.003>0) both positive carry