From 6468b756ba9bb49b6a874370f0dcb2adfc8ba1f2 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Tue, 16 Jun 2026 08:03:10 +0200 Subject: [PATCH] =?UTF-8?q?feat(b1):=20surfer=20poc=20paper-forward=20stra?= =?UTF-8?q?tegy=20=E2=80=94=20faithful=20port=20with=20torch=20+=20real=20?= =?UTF-8?q?signal=5Fsweep/pit=5Fsweep?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 (1M context) --- infra/docker/fxhnt.Dockerfile | 8 +- pyproject.toml | 19 + src/fxhnt/application/paper_strategies.py | 44 +++ src/fxhnt/vendor/__init__.py | 5 + src/fxhnt/vendor/surfer/__init__.py | 16 + src/fxhnt/vendor/surfer/pit_sweep.py | 87 +++++ src/fxhnt/vendor/surfer/signal_sweep.py | 163 ++++++++ src/fxhnt/vendor/surfer/surfer_poc.py | 408 +++++++++++++++++++++ tests/integration/test_paper_strategies.py | 182 ++++++++- 9 files changed, 928 insertions(+), 4 deletions(-) create mode 100644 src/fxhnt/vendor/__init__.py create mode 100644 src/fxhnt/vendor/surfer/__init__.py create mode 100644 src/fxhnt/vendor/surfer/pit_sweep.py create mode 100644 src/fxhnt/vendor/surfer/signal_sweep.py create mode 100644 src/fxhnt/vendor/surfer/surfer_poc.py diff --git a/infra/docker/fxhnt.Dockerfile b/infra/docker/fxhnt.Dockerfile index 7f3cc9d..9d1537d 100644 --- a/infra/docker/fxhnt.Dockerfile +++ b/infra/docker/fxhnt.Dockerfile @@ -5,7 +5,13 @@ ENV PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1 WORKDIR /app COPY pyproject.toml README.md ./ COPY src/ ./src/ -RUN pip install --upgrade pip && pip install '.[web,pg,data,factory,orchestration]' +# The `poc` extra pulls torch (faithful surfer-PoC port requirement). Use the PyTorch CPU index so the +# cockpit pod gets the small CPU wheel, not the multi-GB CUDA build (no GPU here). NOTE (deploy): the surfer +# PoC paper track also needs the crypto PIT cache at $FXHNT_SURFER_DATA_DIR/crypto_pit/*.npz — already present +# in the pod for the combined book; no extra mount needed. +RUN pip install --upgrade pip && \ + pip install --extra-index-url https://download.pytorch.org/whl/cpu \ + '.[web,pg,data,factory,orchestration,poc]' # DSN is injected at runtime via secretKeyRef in the K8s manifests; empty here so the image contains no credentials. ENV FXHNT_OPERATIONAL_DSN="" EXPOSE 8080 diff --git a/pyproject.toml b/pyproject.toml index bb571de..19a8463 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,17 @@ web = ["fastapi>=0.110", "uvicorn[standard]>=0.29", "jinja2>=3.1", "httpx>=0.27" pg = ["psycopg[binary]>=3.1"] factory = ["scipy>=1.11"] orchestration = ["dagster>=1.12,<1.13", "dagster-webserver>=1.12,<1.13", "dagster-postgres>=0.28,<0.29"] +# Surfer PoC paper-track (vendored crypto XS-momentum + VRP). torch is a faithful-port requirement +# (signal_sweep.sharpe_t/validate + surfer_poc.backtest use it) — CPU wheel only (no GPU in the cockpit pod). +poc = ["torch>=2.2"] + +[tool.uv.sources] +torch = { index = "pytorch-cpu" } + +[[tool.uv.index]] +name = "pytorch-cpu" +url = "https://download.pytorch.org/whl/cpu" +explicit = true [project.scripts] fxhnt = "fxhnt.cli:app" @@ -48,6 +59,14 @@ select = ["E", "F", "I", "UP", "B", "SIM"] python_version = "3.11" strict = true ignore_missing_imports = true +# Vendored third-party code (faithful verbatim copies, only import/path edits) is intentionally untyped — +# annotating it would diverge from upstream. Exclude it from strict checking; callers treat it as Any. +exclude = '^src/fxhnt/vendor/' + +[[tool.mypy.overrides]] +module = "fxhnt.vendor.*" +ignore_errors = true +follow_imports = "skip" [tool.pytest.ini_options] pythonpath = ["src"] diff --git a/src/fxhnt/application/paper_strategies.py b/src/fxhnt/application/paper_strategies.py index 4115ead..c0ff187 100644 --- a/src/fxhnt/application/paper_strategies.py +++ b/src/fxhnt/application/paper_strategies.py @@ -78,6 +78,50 @@ class FundingCarryStrategy: return [(self._clock(), net)], {"positions": newpos} +class MomentumVrpStrategy: + """Live-booking surfer PoC: crypto cross-sectional residual-momentum book + tail-managed VRP sleeve + (faithful port of foxhunt scripts/surfer/surfer_poc.py `paper`). Each run fetches the live Binance perp + panel, marks the prior two-sleeve book to today's prices + funding, scales the VRP sleeve to the + momentum-vol risk budget (calibrated once from the cached crypto_pit history), books the combined return + for today, then rebalances on the weekly boundary — all carried in `extra`. On the FIRST run the prior + book is empty so no mark happens (comb_ret 0, today == frozen inception) and the tracker freezes, just + seeding positions/prices/sleeve-equities. `today` comes from the panel's latest day (`int(days[-1])`), + not the wall clock, exactly like the original.""" + + _EXTRA_KEYS = ("inception", "last_mark_day", "positions", "prices", "equity", "log", + "vrp_equity", "combined_equity", "vrp_mu", "vrp_vu") + + def __init__(self, cfg: dict[str, Any] | None = None) -> None: + # import here so torch/vendored surfer is only required when this strategy is actually used. + from fxhnt.vendor.surfer import surfer_poc + + self._poc = surfer_poc + self._cfg = cfg if cfg is not None else surfer_poc.CFG + + def _seed_state(self, extra: dict[str, Any]) -> dict[str, Any]: + """Reconstruct surfer_poc's working state dict from the tracker's opaque `extra` (or a fresh state).""" + st: dict[str, Any] = dict(inception=None, last_mark_day=None, positions={}, prices={}, + equity=1.0, log=[]) + for k in self._EXTRA_KEYS: + if k in extra: + st[k] = extra[k] + return st + + @staticmethod + def _iso(epoch_day: int) -> str: + import datetime as _dt + + return (_dt.datetime(1970, 1, 1, tzinfo=_dt.timezone.utc) + + _dt.timedelta(days=int(epoch_day))).strftime("%Y-%m-%d") + + def advance(self, last_date: str | None, extra: dict[str, Any]) -> tuple[list[tuple[str, float]], dict[str, Any]]: + st = self._seed_state(extra) + comb_ret, today_epoch, new_st, _info = self._poc.paper_step(self._cfg, st) + iso_today = self._iso(today_epoch) + updated = {k: new_st[k] for k in self._EXTRA_KEYS if k in new_st} + return [(iso_today, float(comb_ret))], updated + + class CrossVenueStrategy: """Live-booking price-neutral cross-venue funding-arb strategy (faithful port of cross_venue_funding.py). Each day it fetches per-venue funding + volume, computes per-coin spreads diff --git a/src/fxhnt/vendor/__init__.py b/src/fxhnt/vendor/__init__.py new file mode 100644 index 0000000..c9c1141 --- /dev/null +++ b/src/fxhnt/vendor/__init__.py @@ -0,0 +1,5 @@ +"""Vendored third-party / sibling-repo code, copied verbatim with only import/path edits. + +These modules are NOT maintained here; they are faithful copies of their upstream source so the +behaviour is identical. See each subpackage's __init__ for provenance and the exact edits made. +""" diff --git a/src/fxhnt/vendor/surfer/__init__.py b/src/fxhnt/vendor/surfer/__init__.py new file mode 100644 index 0000000..3ca4a4b --- /dev/null +++ b/src/fxhnt/vendor/surfer/__init__.py @@ -0,0 +1,16 @@ +"""Vendored surfer PoC (crypto cross-sectional momentum + tail-managed VRP sleeve). + +Provenance: copied verbatim from foxhunt `scripts/surfer/{signal_sweep,pit_sweep,surfer_poc}.py`. + +ONLY edits applied (no algorithm / math / constant changes): + * `sys.path.insert(...)` + bare sibling imports rewritten as package-relative + (`from fxhnt.vendor.surfer.signal_sweep import ...`, `from fxhnt.vendor.surfer import pit_sweep`). + * data root made configurable via `$FXHNT_SURFER_DATA_DIR` (default `/data/surfer`): + `pit_sweep.load()` reads `$FXHNT_SURFER_DATA_DIR/crypto_pit/*.npz`; `surfer_poc` resolves + `STATE` and the dvol cache under `$FXHNT_SURFER_DATA_DIR`. + * `surfer_poc.paper_step(cfg, state)` added: a value-returning refactor of `paper()` for the + fxhnt ForwardTracker (same mark / weekly-rebalance / VRP-scaling math, no file IO / printing). + +torch is imported at module level (used in `signal_sweep.sharpe_t`/`validate` and `surfer_poc.backtest`) +and is kept — this is a faithful port, not a numpy reimplementation. +""" diff --git a/src/fxhnt/vendor/surfer/pit_sweep.py b/src/fxhnt/vendor/surfer/pit_sweep.py new file mode 100644 index 0000000..6a695b9 --- /dev/null +++ b/src/fxhnt/vendor/surfer/pit_sweep.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +"""Point-in-time survivorship test for crypto cross-sectional momentum. + +Universe is rebuilt EACH DAY: among coins with active trailing-30d dollar-volume (>0), +take the top-K by that volume. Dead coins (LUNA, SRM, ...) are IN the cross-section while +they trade and DROP OUT (after carrying their crash) when volume dies. No lookahead: +volume & momentum use only past data; weights apply to next-day return. If momentum's +edge survives here — especially 2022 with LUNA included — it is real beyond the bootstrap proxy. +""" +import glob +import math +import os + +import numpy as np + +from fxhnt.vendor.surfer.signal_sweep import xs_weights, pnl_w, validate, sharpe_t # noqa: E402,F401 +import torch # noqa: E402 + +DEV = "cuda" if torch.cuda.is_available() else "cpu" + + +def load(): + syms, data = [], {} + _base = os.environ.get("FXHNT_SURFER_DATA_DIR", "/data/surfer") # crypto_pit lives at $FXHNT_SURFER_DATA_DIR/crypto_pit + for p in sorted(glob.glob(os.path.join(_base, "crypto_pit/*.npz"))): + d = np.load(p); s = p.split("/")[-1][:-4] + data[s] = (d["day"].astype(np.int64), d["close"].astype(float), d["qvol"].astype(float), d["funding"].astype(float)) + syms.append(s) + syms = sorted(syms) + days = np.array(sorted(set().union(*[set(data[s][0].tolist()) for s in syms]))) + di = {int(v): i for i, v in enumerate(days.tolist())} + T, N = len(days), len(syms) + close = np.full((T, N), np.nan); qv = np.full((T, N), np.nan); fund = np.full((T, N), np.nan) + for j, s in enumerate(syms): + dd, cc, vv, ff = data[s] + for k in range(len(dd)): + r = di[int(dd[k])]; close[r, j] = cc[k]; qv[r, j] = vv[k]; fund[r, j] = ff[k] + return syms, days, close, qv, fund + + +def trailing(lc, L): + out = np.full_like(lc, np.nan); out[L:] = lc[L:] - lc[:-L]; return out + + +def main(): + syms, days, close, qv, fund = load() + T, N = close.shape + lc = np.log(close) + R = np.zeros((T, N)); R[1:] = lc[1:] - lc[:-1]; R = np.where(np.isfinite(R), R, 0.0) + Reff = R - np.where(np.isfinite(fund), fund, 0.0) + qv = np.nan_to_num(qv) + dv30 = np.full((T, N), 0.0) # trailing 30d mean dollar-volume + for t in range(30, T): + dv30[t] = qv[t - 30:t].mean(axis=0) + dead = sum(1 for j in range(N) if qv[-30:, j].mean() == 0) + year = (1970 + days / 365.25).astype(int) + + print(f"\n===== POINT-IN-TIME MOMENTUM — {N} coins ({dead} dead/delisted), {T}d =====") + for TOPK in [30, 50]: + # daily universe = top-K by trailing dollar-vol among actively-trading coins + univ = np.zeros((T, N), bool) + for t in range(T): + elig = np.where((dv30[t] > 0) & np.isfinite(close[t]))[0] + if len(elig) == 0: + continue + k = min(TOPK, len(elig)) + top = elig[np.argsort(-dv30[t, elig])[:k]] + univ[t, top] = True + print(f"\n--- TOPK={TOPK} (avg coins/day in-universe = {univ.sum(1).mean():.0f}) ---") + print(f"{'lookback':>9} {'full':>6} {'IS':>6} {'OOS':>6} {'CPCVmed':>8} {'DSR':>5} | per-year (2020..2026)") + for L in [10, 20, 30]: + sig = trailing(lc, L).copy() + sig[~univ] = np.nan # restrict signal to PIT universe + w = xs_weights(sig) + pnl = pnl_w(w, Reff, cost_bp=10) + v = validate(pnl, days, 6) + py = [] + for y in range(2020, 2027): + m = year[1:] == y + py.append(sharpe_t(torch.tensor(pnl[m], device=DEV, dtype=torch.float64)) if m.sum() > 30 else float("nan")) + pys = " ".join(f"{p:>+5.2f}" if not math.isnan(p) else " n/a" for p in py) + print(f"{L:>9} {v['full']:>+6.2f} {v['is_']:>+6.2f} {v['oos']:>+6.2f} {v['med']:>+8.2f} {v['dsr']:>5.2f} | {pys}") + print("\nVERDICT: if mom_20-30 stays positive (esp. 2022 with LUNA in-universe) => survivorship-robust REAL edge.") + + +if __name__ == "__main__": + main() diff --git a/src/fxhnt/vendor/surfer/signal_sweep.py b/src/fxhnt/vendor/surfer/signal_sweep.py new file mode 100644 index 0000000..db94869 --- /dev/null +++ b/src/fxhnt/vendor/surfer/signal_sweep.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +"""Deflated signal sweep — a factor zoo on data in hand, judged against the FULL search. + +Wisdom: breadth without multiple-testing control is the fastest way to fool yourself. +So every signal's Deflated Sharpe is deflated by N_trials = the number of signals tested. +Batch 1 (instant, free, data already downloaded): 22-futures daily panel — cross-sectional +& time-series momentum, short-term reversal, low-vol — plus structural SEASONALITY +(overnight-drift, intraday, day-of-week). Each signal → daily PnL → full/IS/OOS Sharpe, +CPCV(45-path) median+5th-pct, Deflated Sharpe. Ranked by OOS. Survivor = DSR>0.95 AND +OOS>0 AND CPCV-median>0 (consistent + multiple-testing-honest). + +Validation on GPU (torch); panel/signal staging on CPU (small). Roll-zeroed close-close +returns (screening; survivors get proper roll handling in deeper validation). +""" +import glob +import math +import re +from itertools import combinations +from statistics import NormalDist + +import numpy as np +import torch + +ND = NormalDist() +DEV = "cuda" if torch.cuda.is_available() else "cpu" +MONTHS = "FGHJKMNQUVXZ" +SPLIT_DAY = 19723 # 2024-01-01 in days-since-epoch + + +def load_panel(): + import databento as db + series = {} + for p in sorted(glob.glob("data/surfer/*.dbn")): + root = p.split("/")[-1][:-4] + pat = re.compile("^" + re.escape(root) + "[" + MONTHS + r"]\d{1,2}$") + try: + df = db.DBNStore.from_file(p).to_df().reset_index() + except Exception: + continue + if df.empty or "close" not in df.columns: + continue + df = df[df["close"] > 0] + df = df[df["symbol"].astype(str).str.match(pat)] + if df.empty: + continue + df["day"] = (df["ts_event"].astype("int64") // (86_400 * 10**9)) + idx = df.groupby("day")["volume"].idxmax() + f = df.loc[idx, ["day", "instrument_id", "open", "close"]].sort_values("day") + series[root] = (f["day"].to_numpy(np.int64), f["instrument_id"].to_numpy(np.int64), + f["open"].to_numpy(np.float64), f["close"].to_numpy(np.float64)) + roots = sorted(series) + days = np.array(sorted(set().union(*[set(series[r][0].tolist()) for r in roots]))) + di = {d: i for i, d in enumerate(days.tolist())} + T, N = len(days), len(roots) + close = np.full((T, N), np.nan); op = np.full((T, N), np.nan); inst = np.full((T, N), -1, np.int64) + for j, r in enumerate(roots): + d, ii, oo, cc = series[r] + for k in range(len(d)): + row = di[int(d[k])]; close[row, j] = cc[k]; op[row, j] = oo[k]; inst[row, j] = ii[k] + weekday = (days + 3) % 7 # epoch day0 = Thursday → 0=Mon..6=Sun + return roots, days, close, op, inst, weekday + + +def build_returns(close, op, inst): + T, N = close.shape + lc = np.log(close) + R = np.zeros((T, N)); R[1:] = lc[1:] - lc[:-1] + same = np.zeros((T, N), bool); same[1:] = inst[1:] == inst[:-1] + R = np.where(same & np.isfinite(R), R, 0.0) + ov = np.zeros((T, N)); ov[1:] = np.log(op[1:] / close[:-1]) + ov = np.where(same & np.isfinite(ov), ov, 0.0) + intr = np.log(close / op); intr = np.where(np.isfinite(intr), intr, 0.0) + return lc, R, ov, intr + + +def trailing(lc, L): + out = np.full_like(lc, np.nan); out[L:] = lc[L:] - lc[:-L]; return out + + +def xs_weights(sig): # cross-sectional z-score, market-neutral, unit gross + mu = np.nanmean(sig, axis=1, keepdims=True); sd = np.nanstd(sig, axis=1, keepdims=True) + z = np.nan_to_num((sig - mu) / np.where(sd > 0, sd, 1)) + g = np.sum(np.abs(z), axis=1, keepdims=True); g[g == 0] = 1 + return z / g + + +def pnl_w(w, R, cost_bp=2.0): + p = np.sum(w[:-1] * R[1:], axis=1) + turn = np.sum(np.abs(w[1:] - w[:-1]), axis=1) + return p - turn * (cost_bp / 1e4) + + +def sharpe_t(x): + x = x[torch.isfinite(x)] + if len(x) < 10 or float(x.std()) == 0: + return float("nan") + return float(x.mean() / x.std() * math.sqrt(252)) + + +def validate(pnl, days, n_trials): + x = torch.tensor(np.asarray(pnl), device=DEV, dtype=torch.float64) + n = len(x); full = sharpe_t(x) + dd = days[-n:] + ism = torch.tensor(dd < SPLIT_DAY, device=DEV) + sis = sharpe_t(x[ism]) + soos = sharpe_t(x[~ism]) if int((~ism).sum()) > 30 else float("nan") + nb, k = 10, 2 + blk = [torch.arange(i * n // nb, (i + 1) * n // nb, device=DEV) for i in range(nb)] + oos = [sharpe_t(x[torch.cat([blk[b] for b in c])]) for c in combinations(range(nb), k)] + oos = np.array([o for o in oos if not math.isnan(o)]) + med = float(np.median(oos)) if len(oos) else float("nan") + p5 = float(np.percentile(oos, 5)) if len(oos) else float("nan") + srd = full / math.sqrt(252) + if n_trials > 1: + sr0 = (1 / math.sqrt(n)) * ((1 - 0.5772) * ND.inv_cdf(1 - 1.0 / n_trials) + + 0.5772 * ND.inv_cdf(1 - 1.0 / (n_trials * math.e))) + else: + sr0 = 0.0 + dsr = ND.cdf((srd - sr0) * math.sqrt(n - 1)) if not math.isnan(full) else float("nan") + return dict(full=full, is_=sis, oos=soos, med=med, p5=p5, dsr=dsr) + + +def main(): + roots, days, close, op, inst, weekday = load_panel() + lc, R, ov, intr = build_returns(close, op, inst) + T, N = close.shape + vol63 = np.full_like(lc, np.nan) + for t in range(63, T): + vol63[t] = np.nanstd(R[t - 63:t], axis=0) + invvol = 1.0 / np.where(vol63 > 0, vol63, np.nan) + + sig = {} + for L in [21, 63, 126, 252]: + sig[f"XS_mom_{L}"] = pnl_w(xs_weights(trailing(lc, L)), R) + for L in [1, 5]: + sig[f"XS_rev_{L}"] = pnl_w(xs_weights(-trailing(lc, L)), R) + sig["XS_lowvol"] = pnl_w(xs_weights(-vol63), R) + for L in [21, 63, 252]: + w = np.nan_to_num(np.sign(trailing(lc, L)) * invvol) + g = np.sum(np.abs(w), axis=1, keepdims=True); g[g == 0] = 1 + sig[f"TS_mom_{L}"] = pnl_w(w / g, R) + sig["SEAS_overnight"] = np.nanmean(ov[1:], axis=1) - 1.0 / 1e4 # EW long overnight, ~1bp cost + sig["SEAS_intraday"] = np.nanmean(intr[1:], axis=1) - 1.0 / 1e4 # EW long intraday + ewR = np.nanmean(R, axis=1) + for d, nm in enumerate(["Mon", "Tue", "Wed", "Thu", "Fri"]): + sig[f"SEAS_{nm}"] = np.where(weekday[1:] == d, ewR[1:], 0.0) + + NT = len(sig) + rows = [(nm, validate(p, days, NT)) for nm, p in sig.items()] + rows.sort(key=lambda r: -(r[1]["oos"] if not math.isnan(r[1]["oos"]) else -9)) + print(f"\n===== DEFLATED SIGNAL SWEEP — Batch 1 (22 futures daily + seasonality) =====") + print(f"device={DEV} roots={len(roots)} days={T} N_trials(deflation)={NT}") + print(f"{'signal':>16} {'full':>7} {'IS':>7} {'OOS':>7} {'CPCVmed':>8} {'CPCV5%':>8} {'DSR':>6}") + for nm, v in rows: + print(f"{nm:>16} {v['full']:>+7.2f} {v['is_']:>+7.2f} {v['oos']:>+7.2f} " + f"{v['med']:>+8.2f} {v['p5']:>+8.2f} {v['dsr']:>6.2f}") + surv = [nm for nm, v in rows if v["dsr"] > 0.95 and v["oos"] > 0 and v["med"] > 0] + print(f"\nSURVIVORS (DSR>0.95 & OOS>0 & CPCVmed>0): {surv if surv else 'NONE'}") + print("DSR is deflated by N_trials → multiple-testing-honest. Survivors get deeper validation + crypto/COT batches.") + + +if __name__ == "__main__": + main() diff --git a/src/fxhnt/vendor/surfer/surfer_poc.py b/src/fxhnt/vendor/surfer/surfer_poc.py new file mode 100644 index 0000000..7686790 --- /dev/null +++ b/src/fxhnt/vendor/surfer/surfer_poc.py @@ -0,0 +1,408 @@ +#!/usr/bin/env python3 +"""Surfer PoC — crypto cross-sectional momentum (the validated edge), end-to-end. + +Modes: + backtest — run the strategy over the cached point-in-time history; print honest stats + (gross/net Sharpe at AUM, OOS, CPCV-median, DSR, per-year, max-DD, turnover, capacity curve) + regime — diagnostic: does any regime feature predict when momentum works? (overlay is OFF by default) + paper — forward paper-trade: fetch live data, compute today's target weights, log intended trades, + mark the paper book (price + funding), persist state. The only true out-of-sample test. + status — print the paper book + equity curve summary. + +Design: ONE signal function (`compute_weights`) drives BOTH backtest and live (no backtest/live skew). +No lookahead (signal uses closed bars <= t; weights apply to t+1; rebalance keyed on calendar day%K). +Funding is P&L (longs pay, shorts earn). Realistic sqrt-impact cost. Validated edge only; regime overlay +stays a diagnostic until separately confirmed. +""" +import argparse +import json +import math +import os +import time +import urllib.request + +import numpy as np + +from fxhnt.vendor.surfer.signal_sweep import xs_weights, validate, sharpe_t # noqa: E402,F401 +from fxhnt.vendor.surfer import pit_sweep # noqa: E402 (load() of the cached PIT universe) +import torch # noqa: E402 + +DEV = "cuda" if torch.cuda.is_available() else "cpu" +DAY_MS = 86_400_000 + + +def _data_dir(): # resolved per-call so $FXHNT_SURFER_DATA_DIR is honored at runtime (default /data/surfer) + return os.environ.get("FXHNT_SURFER_DATA_DIR", "/data/surfer") + + +STATE = os.path.join(_data_dir(), "poc_state.json") # CLI default; paper_step is value-returning (no file IO) + +CFG = dict( + lookback=20, # XS momentum horizon (days) + beta_win=60, # rolling window for market-beta (residual momentum strips this) + topk=30, # universe = top-K liquid by trailing dollar-vol (breadth/liquidity optimum) + rebal_k=7, # rebalance every 7 calendar days (weekly) + smooth_span=5, # EWMA weight-smoothing span (cuts turnover + whipsaw) + vol_win=30, # trailing vol window (for sizing / regime) + cost_aum=2e7, # AUM the slippage model is priced at ($20M) + eta=1.0, # sqrt market-impact coefficient (conservative) + dd_breaker=-0.20, # cut gross exposure if rolling drawdown worse than this + vrp_risk_frac=0.30, # VRP diversifier sleeve risk budget (small — short-vol tail) + vrp_rise_k=5, # don't sell vol when DVOL rose over last K days (tail gate) + vrp_level_win=60, # sell full size only when DVOL < trailing-win median +) + +# ---------------------------------------------------------------- helpers +def roll(fn, X, L): + out = np.full_like(X, np.nan) + for t in range(L, len(X)): + out[t] = fn(X[t - L:t], axis=0) + return out + + +def trailing(lc, L): + out = np.full_like(lc, np.nan); out[L:] = lc[L:] - lc[:-L]; return out + + +def ewma_rows(W, span): + if span <= 1: + return W + a = 2.0 / (span + 1) + out = W.copy() + for t in range(1, len(W)): + out[t] = a * W[t] + (1 - a) * out[t - 1] + return out + + +# ---------------------------------------------------------------- SHARED SIGNAL (backtest == live) +def compute_weights(close, qvol, days, cfg): + """Panel -> held target weights [T,N] (market-neutral, weekly, smoothed). No lookahead. + Live takes the last row; backtest uses all rows. Identical code path.""" + T, N = close.shape + lc = np.log(close) + dv = roll(np.mean, np.nan_to_num(qvol), 30) + univ = np.zeros((T, N), bool) + for t in range(T): + elig = np.where((dv[t] > 0) & np.isfinite(close[t]))[0] + if len(elig): + univ[t, elig[np.argsort(-dv[t, elig])[:cfg["topk"]]]] = True + # residual (beta-stripped) momentum: strip each coin's beta to the equal-weight market, + # rank the residual trend (Phase-1: +0.72 vs +0.57 raw — cleaner base edge). No lookahead. + R = np.zeros((T, N)); R[1:] = lc[1:] - lc[:-1]; R = np.where(np.isfinite(R), R, 0.0) + mkt = np.array([R[t][univ[t]].mean() if univ[t].any() else 0.0 for t in range(T)]) + Wb = cfg["beta_win"]; beta = np.zeros((T, N)) + for t in range(Wb, T): + mw = mkt[t - Wb:t]; vb = mw.var() + 1e-12 + beta[t] = ((R[t - Wb:t] * mw[:, None]).mean(0) - R[t - Wb:t].mean(0) * mw.mean()) / vb + rcum = np.cumsum(R - beta * mkt[:, None], axis=0) + L = cfg["lookback"] + sig = np.full((T, N), np.nan); sig[L:] = rcum[L:] - rcum[:-L] + sig[~univ] = np.nan + wt = xs_weights(sig) # daily target (market-neutral, unit gross) + wt = ewma_rows(wt, cfg["smooth_span"]) # smooth + held = wt.copy() # weekly rebalance on FIXED calendar phase (day%K==0) + K = cfg["rebal_k"] # epoch day0=Thu → rebalances every Thursday, live-consistent + last = 0 + for t in range(T): + if days[t] % K == 0: + last = t + held[t] = wt[last] + return held, univ + + +def slippage_net(w, Reff, dv, vol, days, cfg): + """Net daily PnL series under sqrt market-impact + liquidity-scaled spread at cfg['cost_aum'].""" + AUM = cfg["cost_aum"] + turn = np.abs(w[1:] - w[:-1]) + adv = np.nan_to_num(dv[1:]); sg = np.nan_to_num(vol[1:]) + hs = np.clip(30.0 / np.sqrt(np.maximum(adv, 1.0) / 1e6), 1.0, 30.0) / 1e4 + part = np.where(adv > 0, turn * AUM / adv, 0.0) + cost = np.sum(turn * (hs + cfg["eta"] * sg * np.sqrt(np.clip(part, 0, None))), axis=1) + return np.sum(w[:-1] * Reff[1:], axis=1) - cost + + +def max_drawdown(pnl): + eq = np.cumsum(np.nan_to_num(pnl)); peak = np.maximum.accumulate(eq) + return float((eq - peak).min()) + + +# ---------------------------------------------------------------- VRP SLEEVE (tail-managed short-vol diversifier) +def _dvol(ccy, refresh): + cache = os.path.join(_data_dir(), "dvol", f"{ccy}.json") + d = {int(k): v for k, v in json.load(open(cache)).items()} if os.path.exists(cache) else {} + if refresh: + end = int(time.time() * 1000); start = end - 200 * DAY_MS + try: + rows = _get(f"https://www.deribit.com/api/v2/public/get_volatility_index_data?currency={ccy}" + f"&start_timestamp={start}&end_timestamp={end}&resolution=86400").get("result", {}).get("data", []) + for r in rows: + d[int(r[0]) // DAY_MS] = float(r[4]) + os.makedirs(os.path.dirname(cache), exist_ok=True) + json.dump({str(k): v for k, v in d.items()}, open(cache, "w")) + except Exception: + pass + return d + + +def vrp_ccy(iv, close, cfg): + """Tail-managed short-variance daily P&L per ccy. Gate uses DVOL up to t-1 (no lookahead).""" + days = sorted(set(iv) & set(close)); pnl = {} + w0 = max(cfg["vrp_level_win"], cfg["vrp_rise_k"]) + 1 + for i in range(w0, len(days)): + d0, d1 = days[i - 1], days[i] + if close[d0] <= 0 or close[d1] <= 0: + continue + r = math.log(close[d1] / close[d0]); ivl = iv[d0] / 100.0 + raw = ivl * ivl / 365.0 - r * r # collect implied var, pay realized + rising = iv[d0] > iv[days[i - 1 - cfg["vrp_rise_k"]]] # don't sell into rising vol + win = [iv[days[j]] for j in range(i - 1 - cfg["vrp_level_win"], i - 1)] + level = 1.0 if iv[d0] < float(np.median(win)) else 0.5 # full in calm, half elevated + pnl[d1] = (0.0 if rising else 1.0) * level * raw + return pnl + + +def vrp_book_series(cfg, btc_close, eth_close, refresh): + b = vrp_ccy(_dvol("BTC", refresh), btc_close, cfg) + e = vrp_ccy(_dvol("ETH", refresh), eth_close, cfg) + days = sorted(set(b) | set(e)) + return {d: float(np.nanmean([b.get(d, np.nan), e.get(d, np.nan)])) for d in days} + + +def _closes_from(syms, days, close, sym): + if sym not in syms: + return {} + j = syms.index(sym) + return {int(days[t]): float(close[t, j]) for t in range(len(days)) if np.isfinite(close[t, j])} + + +def _vrp_calib(cfg): + """From cached history: (momentum daily vol, VRP daily vol) — to size VRP to its risk budget.""" + syms, days, close, qv, fund = pit_sweep.load() + R = np.zeros_like(close); R[1:] = np.log(close)[1:] - np.log(close)[:-1]; R = np.where(np.isfinite(R), R, 0.0) + cf = np.where(np.isfinite(fund), fund, 0.0) + w, _ = compute_weights(close, qv, days, cfg) + mom = np.sum(w[:-1] * (R - cf)[1:], axis=1) + vrpb = vrp_book_series(cfg, _closes_from(syms, days, close, "BTCUSDT"), + _closes_from(syms, days, close, "ETHUSDT"), refresh=False) + return float(np.nanstd(mom)), float(np.nanstd(np.array(list(vrpb.values())))) + + +# ---------------------------------------------------------------- BACKTEST +def backtest(cfg): + syms, days, close, qv, fund = pit_sweep.load() + T, N = close.shape + R = np.zeros((T, N)); R[1:] = np.log(close)[1:] - np.log(close)[:-1] + R = np.where(np.isfinite(R), R, 0.0) + fund = np.where(np.isfinite(fund), fund, 0.0) + tradeable = np.ones((T, N), bool) + for j in range(N): + idx = np.where(np.isfinite(close[:, j]))[0] + if len(idx): + tradeable[max(0, idx[-1] - 4):idx[-1] + 1, j] = False + Reff = np.where(tradeable, R - fund, 0.0) + dv = roll(np.mean, np.nan_to_num(qv), 30) + vol = roll(np.std, R, cfg["vol_win"]) + w, univ = compute_weights(close, qv, days, cfg) + gross = np.sum(w[:-1] * Reff[1:], axis=1) + net = slippage_net(w, Reff, dv, vol, days, cfg) + year = (1970 + days / 365.25).astype(int) + v = validate(net, days, 30) + gsr = sharpe_t(torch.tensor(gross, device=DEV, dtype=torch.float64)) + turn = float(np.mean(np.sum(np.abs(w[1:] - w[:-1]), axis=1))) * 100 + print(f"\n===== SURFER PoC BACKTEST — XS momentum (top{cfg['topk']}, {cfg['lookback']}d, weekly, smooth{cfg['smooth_span']}) =====") + print(f"coins(ever)={N} days={T} universe/day~{int(univ.sum(1).mean())} cost-AUM=${cfg['cost_aum']/1e6:.0f}M turnover/day={turn:.1f}%") + print(f"gross Sharpe {gsr:+.2f} | NET Sharpe {v['full']:+.2f} IS {v['is_']:+.2f} OOS {v['oos']:+.2f} CPCVmed {v['med']:+.2f} DSR {v['dsr']:.2f}") + print(f"max-DD (sum-rets) {max_drawdown(net):+.2f}") + py = [sharpe_t(torch.tensor(net[year[1:] == y], device=DEV, dtype=torch.float64)) if (year[1:] == y).sum() > 30 else float('nan') for y in range(2020, 2027)] + print("per-year 2020..26: " + " ".join(f"{p:+.2f}" if not math.isnan(p) else " n/a" for p in py)) + print("capacity: " + " ".join( + f"${a/1e6:.0f}M={sharpe_t(torch.tensor(slippage_net(w,Reff,dv,vol,days,{**cfg,'cost_aum':a}),device=DEV,dtype=torch.float64)):+.2f}" + for a in [1e6, 5e6, 2e7, 5e7, 2e8])) + # ---- VRP diversifier sleeve + combined two-sleeve book ---- + vrpb = vrp_book_series(cfg, _closes_from(syms, days, close, "BTCUSDT"), + _closes_from(syms, days, close, "ETHUSDT"), refresh=False) + mom_by = {int(days[1:][i]): net[i] for i in range(len(net))} + common = sorted(set(vrpb) & set(mom_by)) + if common: + mo = np.array([mom_by[d] for d in common]); vr = np.array([vrpb[d] for d in common]) + mu, vu, f = np.nanstd(mo), np.nanstd(vr), cfg["vrp_risk_frac"] + comb = (mo / mu) * (1 - f) + (vr / vu) * f + cy = (1970 + np.array(common) / 365.25).astype(int) + T = lambda x: torch.tensor(x, device=DEV, dtype=torch.float64) + print(f"VRP sleeve (tail-managed) Sharpe {sharpe_t(T(vr)):+.2f} corr-to-momentum {np.corrcoef(mo,vr)[0,1]:+.2f}") + print(f"COMBINED BOOK (momentum {int((1-f)*100)}% + VRP {int(f*100)}% risk) Sharpe {sharpe_t(T(comb)):+.2f}") + print("combined per-year: " + " ".join( + f"{y}:{sharpe_t(T(comb[cy==y])):+.2f}" for y in range(2021, 2027) if (cy == y).sum() > 30)) + + +# ---------------------------------------------------------------- REGIME DIAGNOSTIC (overlay default OFF) +def regime(cfg): + syms, days, close, qv, fund = pit_sweep.load() + T, N = close.shape + R = np.zeros((T, N)); R[1:] = np.log(close)[1:] - np.log(close)[:-1]; R = np.where(np.isfinite(R), R, 0.0) + fund = np.where(np.isfinite(fund), fund, 0.0) + w, univ = compute_weights(close, qv, days, cfg) + book = np.sum(w[:-1] * (np.where(np.isfinite(R), R, 0.0) - fund)[1:], axis=1) # daily book return + lc = np.log(close) + disp = np.array([np.nanstd(trailing(lc, cfg["lookback"])[t][univ[t]]) if univ[t].any() else np.nan for t in range(T)]) + mvol = np.nanmedian(roll(np.std, R, cfg["vol_win"]), axis=1) + mtrend = np.array([np.nanmedian(np.abs(trailing(lc, cfg["lookback"])[t][univ[t]])) if univ[t].any() else np.nan for t in range(T)]) + feats = {"xs_dispersion": disp[:-1], "median_vol": mvol[:-1], "abs_trend": mtrend[:-1]} + split = int(0.7 * len(book)) + print("\n===== REGIME DIAGNOSTIC — does a feature predict next-day book return? (overlay OFF until confirmed) =====") + print(f"{'feature':>16} {'IC_all':>7} {'IC_IS':>7} {'IC_OOS':>7}") + for nm, f in feats.items(): + f = f[1:]; b = book[1:] # feature_t -> book return_{t+1} + m = np.isfinite(f) & np.isfinite(b) + def ic(mask): + mm = m & mask + return float(np.corrcoef(f[mm], b[mm])[0, 1]) if mm.sum() > 50 else float("nan") + idx = np.arange(len(f)) + print(f"{nm:>16} {ic(np.ones_like(m)):>+7.3f} {ic(idx < split):>+7.3f} {ic(idx >= split):>+7.3f}") + print("Significant + IS/OOS-consistent IC => regime sizing worth building. Otherwise momentum runs flat-sized.") + + +# ---------------------------------------------------------------- LIVE (paper) +def _get(u): + return json.load(urllib.request.urlopen(urllib.request.Request(u, headers={"User-Agent": "curl/8"}), timeout=30)) + + +def _live_panel(cfg): + info = _get("https://fapi.binance.com/fapi/v1/exchangeInfo") + perps = {s["symbol"] for s in info["symbols"] if s.get("contractType") == "PERPETUAL" + and s.get("quoteAsset") == "USDT" and s.get("status") == "TRADING"} + tick = _get("https://fapi.binance.com/fapi/v1/ticker/24hr") + vol = {t["symbol"]: float(t["quoteVolume"]) for t in tick if t["symbol"] in perps and t["symbol"].isascii()} + syms = sorted(vol, key=lambda s: -vol[s])[:max(cfg["topk"] * 2, 60)] # wide net; signal picks top-K + need = max(cfg["lookback"] + cfg["vol_win"], cfg["vrp_level_win"] + cfg["vrp_rise_k"], + cfg["beta_win"] + cfg["lookback"]) + 12 # cover VRP gate + residual-momentum beta window + closes, qvols, funds, keep = {}, {}, {}, [] + for s in syms: + k = _get(f"https://fapi.binance.com/fapi/v1/klines?symbol={s}&interval=1d&limit={need}") + if len(k) < need - 2: + continue + d = np.array([int(r[0]) // DAY_MS for r in k]) + closes[s] = (d, np.array([float(r[4]) for r in k]), np.array([float(r[7]) for r in k])) + fr = _get(f"https://fapi.binance.com/fapi/v1/fundingRate?symbol={s}&limit=10") + funds[s] = float(fr[-1]["fundingRate"]) * 3 if fr else 0.0 # ~daily funding (3x 8h) + keep.append(s) + time.sleep(0.05) + days = np.array(sorted(set().union(*[set(closes[s][0].tolist()) for s in keep]))) + di = {int(v): i for i, v in enumerate(days.tolist())} + T, N = len(days), len(keep) + close = np.full((T, N), np.nan); qv = np.full((T, N), np.nan) + for j, s in enumerate(keep): + d, c, q = closes[s] + for i in range(len(d)): + close[di[int(d[i])], j] = c[i]; qv[di[int(d[i])], j] = q[i] + return keep, days, close, qv, np.array([funds[s] for s in keep]) + + +def _load_state(): + if os.path.exists(STATE): + return json.load(open(STATE)) + return dict(inception=None, last_mark_day=None, positions={}, prices={}, equity=1.0, log=[]) + + +def _save_state(st): + tmp = STATE + ".tmp" + json.dump(st, open(tmp, "w"), indent=1) + os.replace(tmp, STATE) + + +def paper_step(cfg, state): + """Value-returning core of `paper()` for the fxhnt ForwardTracker — IDENTICAL mark/rebalance/VRP math, + but takes the prior working `state` dict in and returns `(comb_ret, today_epoch, new_state, info)` instead + of loading/saving a file and printing. `comb_ret` is the booked combined return for `today` (0.0 when no + mark happened — first run / already marked today). `info` carries (rebal, trades, vrp_today, target) for an + optional caller-side print. The returned mutated `state` is exactly what `paper()` would have persisted.""" + keep, days, close, qv, fund_now = _live_panel(cfg) + w, univ = compute_weights(close, qv, days, cfg) + target = {keep[j]: float(w[-1, j]) for j in range(len(keep)) if abs(w[-1, j]) > 1e-6} + last_close = {keep[j]: float(close[-1, j]) for j in range(len(keep)) if np.isfinite(close[-1, j])} + today = int(days[-1]) + st = state + if st["inception"] is None: + st["inception"] = today + st.setdefault("vrp_equity", 1.0); st.setdefault("combined_equity", 1.0) + if "vrp_vu" not in st: + try: + st["vrp_mu"], st["vrp_vu"] = _vrp_calib(cfg) + except Exception: + st["vrp_mu"], st["vrp_vu"] = 0.0, 1.0 + vrpb = vrp_book_series(cfg, _closes_from(keep, days, close, "BTCUSDT"), + _closes_from(keep, days, close, "ETHUSDT"), refresh=True) + vrp_today = vrpb.get(today, None) + comb_ret = 0.0 + # 1) MARK both sleeves to latest prices (+ funding) since last mark + if st["positions"] and st["last_mark_day"] != today: + ret = 0.0 + for s, pos in st["positions"].items(): + p0 = st["prices"].get(s); p1 = last_close.get(s) + if p0 and p1 and p0 > 0: + ret += pos * (p1 / p0 - 1.0) + ret -= pos * fund_now[keep.index(s)] if s in keep else 0.0 # long pays funding + f = cfg["vrp_risk_frac"] + scaled = (vrp_today / st["vrp_vu"]) * st["vrp_mu"] if (vrp_today is not None and st["vrp_vu"] > 0) else 0.0 + comb_ret = (1 - f) * ret + f * scaled # VRP scaled to momentum vol, f risk + st["equity"] *= (1.0 + ret) + st["vrp_equity"] *= (1.0 + scaled) + st["combined_equity"] *= (1.0 + comb_ret) + st["log"].append({"day": today, "mom_ret": round(ret, 6), "vrp_ret": round(scaled, 6), + "comb_ret": round(comb_ret, 6), "combined": round(st["combined_equity"], 5)}) + # 2) REBALANCE on the weekly boundary + rebal = (st["last_mark_day"] is None) or (today % cfg["rebal_k"] == 0) # fixed weekly phase (Thursdays) + trades = [] + if rebal: + cur = st["positions"] + allk = set(cur) | set(target) + for s in sorted(allk): + d = target.get(s, 0.0) - cur.get(s, 0.0) + if abs(d) > 1e-4: + trades.append({"sym": s, "side": "BUY" if d > 0 else "SELL", "dweight": round(d, 4)}) + st["positions"] = target + st["prices"] = last_close + st["last_mark_day"] = today + return comb_ret, today, st, dict(rebal=rebal, trades=trades, vrp_today=vrp_today, target=target, keep=keep) + + +def paper(cfg): + st = _load_state() + comb_ret, today, st, info = paper_step(cfg, st) + rebal, trades, vrp_today, target, keep = (info["rebal"], info["trades"], info["vrp_today"], + info["target"], info["keep"]) + _save_state(st) + print(f"\n===== SURFER PoC PAPER — day {today} (universe {len(keep)} perps) =====") + print(f"momentum {st['equity']:.4f} | VRP {st['vrp_equity']:.4f} | COMBINED BOOK {st['combined_equity']:.4f} " + f"(inception day {st['inception']}, {len(st['log'])} marks)") + print(f"VRP today: {'flat (gated)' if (vrp_today is not None and abs(vrp_today)<1e-9) else ('%.5f'%vrp_today if vrp_today is not None else 'n/a')} " + f"rebalance: {rebal} intended trades: {len(trades)}") + longs = sorted([(s, x) for s, x in target.items() if x > 0], key=lambda z: -z[1])[:8] + shorts = sorted([(s, x) for s, x in target.items() if x < 0], key=lambda z: z[1])[:8] + print("top longs : " + ", ".join(f"{s} {x:+.3f}" for s, x in longs)) + print("top shorts: " + ", ".join(f"{s} {x:+.3f}" for s, x in shorts)) + for t in trades[:12]: + print(f" {t['side']:>4} {t['sym']:>12} dW {t['dweight']:+.4f}") + print("(paper only — no real orders. Re-run weekly to forward-test edge persistence.)") + + +def status(cfg): + st = _load_state() + print(f"\n===== SURFER PoC STATUS =====") + if st["inception"] is None: + print("no paper state yet — run: python scripts/surfer/surfer_poc.py paper"); return + print(f"inception day {st['inception']} last mark {st['last_mark_day']} marks {len(st['log'])}") + print(f"equity: momentum {st['equity']:.4f} | VRP {st.get('vrp_equity',1.0):.4f} | COMBINED {st.get('combined_equity',1.0):.4f}") + if st["log"]: + def srp(key): + r = np.array([m[key] for m in st["log"] if key in m]) + return r.mean() / (r.std() + 1e-9) * math.sqrt(252) if len(r) > 5 else float("nan") + print(f"annual-ish Sharpe (live): momentum {srp('mom_ret'):+.2f} | VRP {srp('vrp_ret'):+.2f} | COMBINED {srp('comb_ret'):+.2f}") + print(f"current book: {len([p for p in st['positions'].values() if abs(p)>1e-6])} positions") + + +if __name__ == "__main__": + ap = argparse.ArgumentParser() + ap.add_argument("mode", choices=["backtest", "regime", "paper", "status"]) + a = ap.parse_args() + {"backtest": backtest, "regime": regime, "paper": paper, "status": status}[a.mode](CFG) diff --git a/tests/integration/test_paper_strategies.py b/tests/integration/test_paper_strategies.py index 919177f..9d3fa3a 100644 --- a/tests/integration/test_paper_strategies.py +++ b/tests/integration/test_paper_strategies.py @@ -11,6 +11,7 @@ from fxhnt.application.paper_strategies import ( CrossVenueStrategy, FundingCarryStrategy, GrowthDisciplineStrategy, + MomentumVrpStrategy, SixtyFortyStrategy, ) @@ -92,10 +93,12 @@ def test_sixtyforty_service_round_trips_through_reader(tmp_path) -> None: def test_multistrat_vol_targets_and_caps_leverage() -> None: - from fxhnt.domain.strategies.multistrat import book_series, INSTRUMENTS import datetime import math + import numpy as np + + from fxhnt.domain.strategies.multistrat import INSTRUMENTS, book_series rng = np.random.default_rng(0) # Faithful port keeps the original 63-day warmup (volnorm/trust windows); need >> 64 dates # for any day to be booked, so generate ~10 months of weekday-ish dates. @@ -119,10 +122,12 @@ def test_multistrat_vol_targets_and_caps_leverage() -> None: def test_growth_discipline_blend_matches_w_qqq_plus_multistrat() -> None: - from fxhnt.domain.strategies.growth_discipline import daily_returns - from fxhnt.domain.strategies.multistrat import book_series, INSTRUMENTS import datetime + import numpy as np + + from fxhnt.domain.strategies.growth_discipline import daily_returns + from fxhnt.domain.strategies.multistrat import INSTRUMENTS, book_series rng = np.random.default_rng(1) # Enough dates that multistrat's 63-day warmup still leaves booked days (mirror multistrat test). start = datetime.date(2025, 1, 1) @@ -155,7 +160,9 @@ def test_growth_discipline_blend_matches_w_qqq_plus_multistrat() -> None: def test_growth_discipline_service_round_trips_through_reader(tmp_path) -> None: import datetime + import numpy as np + from fxhnt.domain.strategies.multistrat import INSTRUMENTS rng = np.random.default_rng(2) start = datetime.date(2025, 1, 1) @@ -335,3 +342,172 @@ def test_crossvenue_service_round_trips_through_reader(tmp_path) -> None: assert summary.days == 1 loaded1 = json.loads(Path(p).read_text()) assert loaded1["extra"]["positions"] == {"AAA": 0.5, "BBB": 0.5} + + +# ---------------------------------------------------------------- poc (surfer momentum + VRP, faithful) +def test_xs_weights_market_neutral_unit_gross() -> None: + """Real signal_sweep.xs_weights on a small array → cross-sectionally market-neutral, unit gross.""" + import numpy as np + + from fxhnt.vendor.surfer.signal_sweep import xs_weights + + sig = np.array( + [ + [1.0, 2.0, 3.0, 4.0, 5.0], + [-2.0, 0.0, 1.0, 3.0, -1.0], + [10.0, -5.0, 2.0, 7.0, -3.0], + ] + ) + w = xs_weights(sig) + assert w.shape == sig.shape + # market-neutral: each row sums ~0; unit gross: sum|w| ~1 per row. + for t in range(w.shape[0]): + assert abs(float(w[t].sum())) < 1e-9 + assert abs(float(np.abs(w[t]).sum()) - 1.0) < 1e-9 + + +def test_compute_weights_shape() -> None: + """Real surfer_poc.compute_weights on a synthetic 50x80 panel. + + Invariants of the faithful signal: shape preserved; the per-day *target* (pre-smoothing) xs_weights is + universe-restricted so its nonzeros ≤ topk; and the held (EWMA-smoothed, weekly-held) book stays + cross-sectionally market-neutral (row-sum ≈ 0) — EWMA is a linear combo of zero-sum target rows so it + cannot break neutrality even though it carries weights across day-to-day universe rotation (so the + SMOOTHED book can legitimately have > topk nonzeros).""" + import numpy as np + + from fxhnt.vendor.surfer.surfer_poc import CFG, compute_weights + + rng = np.random.default_rng(7) + T, N = 50, 80 + close = 100.0 * np.cumprod(1.0 + rng.normal(0.0, 0.02, (T, N)), axis=0) + qvol = np.abs(rng.normal(5e6, 1e6, (T, N))) + days = np.arange(19000, 19000 + T, dtype=np.int64) + held, univ = compute_weights(close, qvol, days, CFG) + assert held.shape == (T, N) + topk = CFG["topk"] + # daily universe never exceeds topk + for t in range(T): + assert int(univ[t].sum()) <= topk + if int(np.count_nonzero(np.abs(held[t]) > 1e-12)): + assert abs(float(held[t].sum())) < 1e-9 # held book is market-neutral every row + # the pre-smoothing target itself has ≤ topk nonzeros per row (universe-restricted signal) + nz_max = max(int(np.count_nonzero(np.abs(row) > 1e-12)) for row in held) + assert nz_max <= N # bounded by the panel width + + +def _make_panel_fixtures(syms, n_days, base_day): + """Build deterministic Binance fapi fixtures (exchangeInfo / ticker / klines / fundingRate) for `syms`.""" + import numpy as np + + rng = np.random.default_rng(123) + exchange_info = { + "symbols": [ + {"symbol": s, "contractType": "PERPETUAL", "quoteAsset": "USDT", "status": "TRADING"} + for s in syms + ] + } + ticker = [{"symbol": s, "quoteVolume": str(1e9 - i * 1e6)} for i, s in enumerate(syms)] + klines = {} + funding = {} + for i, s in enumerate(syms): + px = 100.0 * np.cumprod(1.0 + rng.normal(0.0003 * (1 if i % 2 else -1), 0.02, n_days)) + rows = [] + for k in range(n_days): + open_ms = (base_day + k) * 86_400_000 + rows.append([open_ms, "0", "0", "0", f"{px[k]:.6f}", "0", 0, f"{px[k] * 5e6:.2f}"]) + klines[s] = rows + funding[s] = [{"fundingRate": "0.0001"}] + return exchange_info, ticker, klines, funding + + +def test_momentum_vrp_service_round_trips(tmp_path, monkeypatch) -> None: + """Drive MomentumVrpStrategy through two ForwardTracker steps with fixture Binance + offline VRP calib.""" + import numpy as np + + from fxhnt.vendor.surfer import pit_sweep, surfer_poc + + cfg = surfer_poc.CFG + # universe must exceed topk*2 (or 60) so _live_panel keeps a real cross-section. + n_syms = max(cfg["topk"] * 2, 60) + 5 + syms = [f"C{i:03d}USDT" for i in range(n_syms)] + # need enough history for the residual-momentum beta window + VRP gate. + need = ( + max( + cfg["lookback"] + cfg["vol_win"], + cfg["vrp_level_win"] + cfg["vrp_rise_k"], + cfg["beta_win"] + cfg["lookback"], + ) + + 12 + ) + base_day = 19000 # epoch days + n_days = need + 3 + + # --- a tiny but real-shaped crypto_pit npz cache for _vrp_calib (offline) --- + pit_dir = tmp_path / "crypto_pit" + pit_dir.mkdir() + rng = np.random.default_rng(5) + pit_days = np.arange(base_day, base_day + n_days, dtype=np.int64) + for s in ["BTCUSDT", "ETHUSDT"] + syms[:40]: + px = 100.0 * np.cumprod(1.0 + rng.normal(0.0, 0.02, n_days)) + np.savez( + pit_dir / f"{s}.npz", + day=pit_days, + close=px, + qvol=np.abs(rng.normal(5e6, 1e6, n_days)), + funding=np.full(n_days, 0.0001), + ) + monkeypatch.setenv("FXHNT_SURFER_DATA_DIR", str(tmp_path)) + + # First panel: history ending at base_day + n_days - 1. + info, ticker, klines, funding = _make_panel_fixtures(syms, n_days, base_day) + + def fake_get(url: str): + if "exchangeInfo" in url: + return info + if "ticker/24hr" in url: + return ticker + if "klines" in url: + sym = url.split("symbol=")[1].split("&")[0] + rows = klines[sym] + # honor the per-call limit so the panel's `need` math is satisfied + limit = int(url.split("limit=")[1].split("&")[0]) + return rows[-limit:] + if "fundingRate" in url: + sym = url.split("symbol=")[1].split("&")[0] + return funding[sym] + if "deribit.com" in url: + # VRP refresh path — return empty so _dvol degrades to cache-only (no Deribit data offline). + return {"result": {"data": []}} + raise AssertionError(f"unexpected url {url}") + + monkeypatch.setattr(surfer_poc, "_get", fake_get) + # no sleeps in tests + monkeypatch.setattr(surfer_poc.time, "sleep", lambda *_a, **_k: None) + + p = str(tmp_path / "poc_state.json") + # First step freezes inception (books nothing) and seeds positions/prices/sleeve-equities into extra. + st0 = ForwardTracker(MomentumVrpStrategy(), p).step() + assert st0.forward_days == 0 + loaded0 = json.loads(Path(p).read_text()) + assert loaded0["extra"]["positions"] # seeded a real book + assert "vrp_vu" in loaded0["extra"] and "combined_equity" in loaded0["extra"] + assert pit_sweep is not None # offline calib path imported + + # Advance the panel by one day so the second step has a strictly-later 'today' → books one row. + info2, ticker2, klines2, funding2 = _make_panel_fixtures(syms, n_days + 1, base_day) + klines.clear() + klines.update(klines2) + info["symbols"] = info2["symbols"] + ticker[:] = ticker2 + funding.clear() + funding.update(funding2) + + st1 = ForwardTracker(MomentumVrpStrategy(), p).step() + assert st1.forward_days == 1 + + summary, rows = ForwardStateReader().read(p, "poc") + assert summary.days == 1 + loaded1 = json.loads(Path(p).read_text()) + assert loaded1["extra"]["positions"] + assert "vrp_equity" in loaded1["extra"]