From 9e9fe2a13cc6faedd81606d968767e9086cb15d1 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Fri, 19 Jun 2026 07:45:45 +0200 Subject: [PATCH] feat(xsfunding): live-booking forward strategy (delta-neutral funding+basis, executable) Co-Authored-By: Claude Opus 4.8 (1M context) --- src/fxhnt/application/xsfunding_strategy.py | 60 ++++++++++++++++++++ tests/integration/test_xsfunding_strategy.py | 46 +++++++++++++++ 2 files changed, 106 insertions(+) create mode 100644 src/fxhnt/application/xsfunding_strategy.py create mode 100644 tests/integration/test_xsfunding_strategy.py diff --git a/src/fxhnt/application/xsfunding_strategy.py b/src/fxhnt/application/xsfunding_strategy.py new file mode 100644 index 0000000..876cc4b --- /dev/null +++ b/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 diff --git a/tests/integration/test_xsfunding_strategy.py b/tests/integration/test_xsfunding_strategy.py new file mode 100644 index 0000000..4377c4d --- /dev/null +++ b/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