diff --git a/src/fxhnt/application/crypto_factor_runner.py b/src/fxhnt/application/crypto_factor_runner.py new file mode 100644 index 0000000..693fbfc --- /dev/null +++ b/src/fxhnt/application/crypto_factor_runner.py @@ -0,0 +1,169 @@ +"""Walk-forward cross-sectional crypto PRICE-factor backtest over the crypto_pit panel. +Daily rebalance, PIT universe, equal-weight construction by a pluggable price/volume score, +booking the PRICE return of the held weights minus a liquidity-scaled turnover cost. +Produces a daily net-return series per construction mode. + +MIRRORS funding_backtest_runner: the only differences are the pluggable `score_fn` dimension +(callers bind lookback/window via functools.partial) and PRICE-return booking (reusing +`daily_returns_for_weights`) instead of signed funding carry.""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Callable + +import numpy as np + +from fxhnt.domain.backtest import compute_stats +from fxhnt.domain.cross_sectional_funding import Panel, construction_weights, eligible_asof +from fxhnt.domain.equity_backtest import daily_returns_for_weights +from fxhnt.domain.gauntlet.core import evaluate, per_period_sharpe + +_PERIODS_PER_YEAR = 365 # crypto trades daily, all year + +_MODES = ("long_tilt", "market_neutral") + +# score_fn(panel, eligible, asof_day) -> {symbol: score}; lookback/window bound via partial. +ScoreFn = Callable[[Panel, list[str], int], dict[str, float]] + + +@dataclass(frozen=True) +class FactorRunResult: + returns_by_mode: dict[str, np.ndarray] + n_names_avg: float + n_rebalances: int # active rebalances actually booked (= len of each mode's series) + + +def _close_panel(panel: Panel, subset: set[str]) -> dict[str, dict[int, float]]: + """Build a close-only panel {s:{day:close}} for `subset` from the (close, qvol, funding) panel.""" + return {s: {d: panel[s][d][0] for d in panel[s]} for s in subset} + + +class CryptoFactorRunner: + def __init__(self, panel: Panel, score_fn: ScoreFn, *, + min_qvol: float = 1e6, min_history: int = 30, + quantile: float = 0.2, cost_bps: float = 8.0, + slip_coef: float = 0.0005, slip_cap_bps: float = 50.0, + liq_ref: float = 1e6) -> None: + self._p = panel + self._score_fn = score_fn + self._minq, self._minh, self._q = min_qvol, min_history, quantile + self._cost = cost_bps / 1e4 + self._slip_coef, self._slip_cap = slip_coef, slip_cap_bps / 1e4 + self._liq_ref = liq_ref + + 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 _last_qvol(self, sym: str, day: int) -> float: + """Most recent qvol on/before `day` for `sym` (held coins that dropped out still have history).""" + series = self._p.get(sym, {}) + days = [x for x in series if x <= day] + return series[max(days)][1] if days else 0.0 + + def run(self) -> FactorRunResult: + all_days = sorted({d for s in self._p.values() for d in s}) + closes = _close_panel(self._p, set(self._p)) + 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 = self._score_fn(self._p, elig, d) + held = set().union(*(prev_w[m] for m in _MODES)) if any(prev_w[m] for m in _MODES) else set() + cost = {s: self._coin_cost(self._last_qvol(s, d)) for s in set(elig) | held} + names_seen += len(elig) + n_rebals += 1 + for m in _MODES: + w = construction_weights(scores, m, quantile=self._q) + series[m].append(_book(w, prev_w[m], closes, d, nxt, cost)) + prev_w[m] = w + return FactorRunResult( + 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, + n_rebalances=n_rebals, + ) + + +def _book(w: dict[str, float], prev_w: dict[str, float], + closes: dict[str, dict[int, float]], d: int, nxt: int, + cost: dict[str, float]) -> float: + """PRICE return the NEW book `w` earns over [d, nxt] (via daily_returns_for_weights, borrow=0) + minus the round-trip rebalance cost charged on turnover between the prior book and the new one.""" + rets = daily_returns_for_weights(w, closes, [d, nxt], borrow_daily=0.0) + realized = rets[0] if rets else 0.0 + 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 + + +def _json_floats(d: dict[str, Any]) -> dict[str, Any]: + """Coerce model_dump scalars to JSON-native: numpy → python float, NaN/inf → 0.0.""" + out: dict[str, Any] = {} + for k, v in d.items(): + if isinstance(v, (np.floating, float)): + fv = float(v) + out[k] = fv if np.isfinite(fv) else 0.0 + elif isinstance(v, (np.integer,)): + out[k] = int(v) + else: + out[k] = v + return out + + +def evaluate_factor_matrix(panel: Panel, factors: dict[str, ScoreFn], *, + floors: list[float], 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 each factor's runner at every liquidity floor, gauntlet every (factor, mode, floor) cell. + n_trials = total cells tested (multiple-testing correction). + + Measures CROSS-SECTIONAL PRICE-FACTOR returns: each cell's return series is the price return of + the equal-weight construction (long_tilt / market_neutral) minus turnover cost. MIRRORS + evaluate_funding_matrix — the only differences are the score_fn dimension and price-return booking.""" + runs = {(name, f): CryptoFactorRunner(panel, fn, min_qvol=f, quantile=quantile, + cost_bps=cost_bps, slip_coef=slip_coef).run() + for name, fn in factors.items() for f in floors} + series_by_cell = {(name, m, f): runs[(name, f)].returns_by_mode[m] + for name in factors for f in floors for m in _MODES} + n_trials = len(series_by_cell) + 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, + "caveats": { + "price_factor": ("stats.sharpe/cagr/maxDD measure CROSS-SECTIONAL PRICE-FACTOR " + "returns (equal-weight long/long-short minus turnover cost). " + "The illiquidity factor's long leg is thin coins — its net edge " + "is COST-FRAGILE; raise cost_bps/slip_coef to stress it."), + "sharpe_annualization": ("stats.sharpe uses 365/yr (crypto); verdict is_sharpe/" + "oos_sharpe use the gauntlet's 252/yr default. PASS/FAIL is " + "internally consistent; do not compare the two Sharpes numerically."), + }, + "settings": {"floors": floors, "quantile": quantile, "cost_bps": cost_bps, + "slip_coef": slip_coef, "oos_fraction": oos_fraction, + "factors": sorted(factors)}} + for (name, m, f), r in series_by_cell.items(): + stats = compute_stats(r, periods_per_year=_PERIODS_PER_YEAR) + key = f"{name}/{m}@{f:.0e}" + if len(r) < 3: + out["cells"][key] = {"factor": name, "mode": m, "floor": float(f), + "stats": _json_floats(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=None) # price factors have no a-priori structural premium + out["cells"][key] = {"factor": name, "mode": m, "floor": float(f), + "stats": _json_floats(stats.model_dump()), + "verdict": _json_floats(v.model_dump())} + return out diff --git a/src/fxhnt/domain/crypto_factors.py b/src/fxhnt/domain/crypto_factors.py new file mode 100644 index 0000000..c5cc08c --- /dev/null +++ b/src/fxhnt/domain/crypto_factors.py @@ -0,0 +1,51 @@ +"""Pure cross-sectional crypto PRICE/VOLUME factor scores on the crypto_pit panel +{sym:{day:(close, qvol, funding)}}. Each returns {symbol: score} aligned to `eligible` +(robust_z across peers; higher score = stronger long). No I/O.""" +from __future__ import annotations + +from fxhnt.domain.cross_sectional_funding import Panel +from fxhnt.domain.strategies.equity_factor import realized_vol_from_closes, robust_z + + +def xs_reversal_score(panel: Panel, eligible: list[str], asof_day: int, lookback: int = 3) -> dict[str, float]: + """Short-term reversal: long recent LOSERS. score = robust_z(-recent_return). + recent_return = close[asof]/close[asof-lookback]-1 using the nearest bars <= asof.""" + vals: list[float | None] = [] + for s in eligible: + days = sorted(d for d in panel[s] if d <= asof_day) + # need a bar at/near asof and one ~lookback before + if len(days) < 2: + vals.append(None) + continue + c_now = panel[s][days[-1]][0] + prior = [d for d in days if d <= asof_day - lookback] + if not prior or c_now <= 0: + vals.append(None) + continue + c_then = panel[s][prior[-1]][0] + vals.append(-(c_now / c_then - 1.0) if c_then > 0 else None) + z = robust_z(vals) + return {eligible[i]: z[i] for i in range(len(eligible))} + + +def xs_lowvol_score(panel: Panel, eligible: list[str], asof_day: int, window: int = 30) -> dict[str, float]: + """Low-volatility: long LOW-vol coins. score = robust_z(-trailing_realized_vol).""" + vals: list[float | None] = [] + for s in eligible: + closes = [panel[s][d][0] for d in sorted(panel[s]) if d <= asof_day] + v = realized_vol_from_closes(closes, window=window) + vals.append(-v if v is not None else None) + z = robust_z(vals) + return {eligible[i]: z[i] for i in range(len(eligible))} + + +def xs_illiquidity_score(panel: Panel, eligible: list[str], asof_day: int, window: int = 30) -> dict[str, float]: + """Illiquidity premium: long LOW dollar-volume coins. score = robust_z(-trailing_mean_qvol). + (Flagged cost-fragile — the long leg is thin coins.)""" + lo = asof_day - window + vals: list[float | None] = [] + for s in eligible: + qv = [panel[s][d][1] for d in panel[s] if lo < d <= asof_day] + vals.append(-(sum(qv) / len(qv)) if qv else None) + z = robust_z(vals) + return {eligible[i]: z[i] for i in range(len(eligible))} diff --git a/tests/integration/test_crypto_factor_runner.py b/tests/integration/test_crypto_factor_runner.py new file mode 100644 index 0000000..c75bab4 --- /dev/null +++ b/tests/integration/test_crypto_factor_runner.py @@ -0,0 +1,122 @@ +"""Integration tests for the cross-sectional crypto PRICE-factor walk-forward runner + +gauntlet matrix. Synthetic panels with a built-in edge; no I/O, no live network.""" +import json +from functools import partial + +import numpy as np + +from fxhnt.application.crypto_factor_runner import ( + CryptoFactorRunner, + FactorRunResult, + evaluate_factor_matrix, +) +from fxhnt.domain.crypto_factors import ( + xs_illiquidity_score, + xs_lowvol_score, + xs_reversal_score, +) + +Panel = dict[str, dict[int, tuple[float, float, float]]] + + +def _lowvol_edge_panel(n_days: int = 120) -> Panel: + """Low-vol coins drift UP smoothly; high-vol coins oscillate around flat with no drift. + The low-vol factor's long_tilt should therefore make money.""" + panel: Panel = {} + # calm uptrenders: tiny noise, positive drift + for sym, drift in {"CALM1": 0.004, "CALM2": 0.0035, "CALM3": 0.003}.items(): + closes = [] + p = 100.0 + for i in range(n_days): + p *= (1.0 + drift + 0.0003 * ((-1) ** i)) + closes.append(p) + panel[sym] = {1000 + i: (closes[i], 5e6, 0.0) for i in range(n_days)} + # wild flat coins: big oscillation, no net drift + for sym, amp in {"WILD1": 8.0, "WILD2": 10.0, "WILD3": 12.0}.items(): + panel[sym] = {1000 + i: (100.0 + amp * ((-1) ** i), 5e6, 0.0) for i in range(n_days)} + return panel + + +def _generic_panel(n_days: int = 120) -> Panel: + panel: Panel = {} + rng = np.random.default_rng(7) + for k, sym in enumerate(["A", "B", "C", "D", "E", "F"]): + p = 100.0 + closes = [] + for _ in range(n_days): + p *= (1.0 + float(rng.normal(0.0005 * k, 0.01))) + closes.append(max(1.0, p)) + qv = 1e5 * (k + 1) ** 2 + panel[sym] = {1000 + i: (closes[i], qv, 0.0) for i in range(n_days)} + return panel + + +def test_lowvol_factor_long_tilt_has_positive_edge(): + runner = CryptoFactorRunner( + _lowvol_edge_panel(), + partial(xs_lowvol_score, window=30), + min_qvol=1e6, min_history=20, quantile=0.5, cost_bps=0.0, slip_coef=0.0, + ) + res = runner.run() + assert isinstance(res, FactorRunResult) + assert set(res.returns_by_mode) == {"long_tilt", "market_neutral"} + for series in res.returns_by_mode.values(): + assert len(series) > 30 and np.isfinite(series).all() + # the built-in low-vol edge => the long_tilt series is net positive + assert float(np.sum(res.returns_by_mode["long_tilt"])) > 0.0 + assert res.n_rebalances == len(res.returns_by_mode["long_tilt"]) + assert res.n_names_avg > 0.0 + + +def test_runner_books_price_returns_not_funding(): + # funding is nonzero but the runner must book PRICE returns; a flat-price panel with + # nonzero funding must produce ~zero return (proving it is NOT reading funding). + panel: Panel = {} + for sym in ["A", "B", "C", "D"]: + panel[sym] = {1000 + i: (100.0, 5e6, 0.05) for i in range(60)} # huge funding, flat price + res = CryptoFactorRunner( + panel, partial(xs_lowvol_score, window=30), + min_qvol=1e6, min_history=20, quantile=0.5, cost_bps=0.0, slip_coef=0.0, + ).run() + assert abs(float(np.sum(res.returns_by_mode["long_tilt"]))) < 1e-9 + + +def test_runner_deterministic(): + mk = lambda: CryptoFactorRunner( + _lowvol_edge_panel(), partial(xs_lowvol_score, window=30), + min_qvol=1e6, min_history=20, quantile=0.5, cost_bps=8.0, slip_coef=0.0005, + ).run().returns_by_mode["long_tilt"] + assert np.array_equal(mk(), mk()) + + +def test_evaluate_factor_matrix_runs_all_cells(): + factors = { + "reversal": partial(xs_reversal_score, lookback=3), + "lowvol": partial(xs_lowvol_score, window=30), + "illiquidity": partial(xs_illiquidity_score, window=30), + } + report = evaluate_factor_matrix( + _generic_panel(), factors, floors=[1e4, 1e6], quantile=0.5, + cost_bps=8.0, slip_coef=0.0, oos_fraction=0.4, + ) + # 3 factors x 2 floors x 2 modes = 12 cells + assert len(report["cells"]) == 12 + assert report["n_trials"] == 12 + for cell in report["cells"].values(): + assert {"factor", "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"] == 12 + json.dumps(report) # must be JSON-serializable (no NaN/np types) + + +def test_evaluate_factor_matrix_n_trials_equals_total_cells(): + factors = {"lowvol": partial(xs_lowvol_score, window=30)} + report = evaluate_factor_matrix( + _generic_panel(), factors, floors=[1e4], quantile=0.5, + cost_bps=8.0, slip_coef=0.0, oos_fraction=0.4, + ) + # 1 factor x 1 floor x 2 modes = 2 cells + assert report["n_trials"] == 2 + assert len(report["cells"]) == 2 + json.dumps(report) diff --git a/tests/unit/test_crypto_factors.py b/tests/unit/test_crypto_factors.py new file mode 100644 index 0000000..fea5f35 --- /dev/null +++ b/tests/unit/test_crypto_factors.py @@ -0,0 +1,118 @@ +"""Unit tests for the pure cross-sectional crypto PRICE/VOLUME factor scores. +Synthetic panels only — no I/O, no live network.""" +import math + +from fxhnt.domain.crypto_factors import ( + xs_illiquidity_score, + xs_lowvol_score, + xs_reversal_score, +) + +Panel = dict[str, dict[int, tuple[float, float, float]]] + + +def _flat_panel(closes_by_sym: dict[str, list[float]], qvol: float = 5e6, + funding: float = 0.0, start: int = 1000) -> Panel: + """Build a Panel {sym:{day:(close, qvol, funding)}} from per-symbol close lists.""" + panel: Panel = {} + for sym, closes in closes_by_sym.items(): + panel[sym] = {start + i: (c, qvol, funding) for i, c in enumerate(closes)} + return panel + + +def test_reversal_ranks_recent_losers_highest(): + # WIN rallied recently, LOSE dropped, FLAT unchanged. Reversal longs the loser. + panel = _flat_panel({ + "WIN": [100.0] * 10 + [100.0, 110.0, 120.0, 130.0], + "LOSE": [100.0] * 10 + [100.0, 90.0, 80.0, 70.0], + "FLAT": [100.0] * 14, + "MID": [100.0] * 10 + [100.0, 102.0, 104.0, 106.0], + }) + asof = 1013 + elig = ["WIN", "LOSE", "FLAT", "MID"] + scores = xs_reversal_score(panel, elig, asof, lookback=3) + assert set(scores) == set(elig) + # the recent loser scores highest, the recent winner lowest + assert scores["LOSE"] == max(scores.values()) + assert scores["WIN"] == min(scores.values()) + + +def test_lowvol_ranks_low_vol_coins_highest(): + # CALM has tiny daily moves, WILD oscillates hard. Low-vol longs CALM. + calm = [100.0 + 0.05 * ((-1) ** i) for i in range(60)] + wild = [100.0 + 10.0 * ((-1) ** i) for i in range(60)] + medium = [100.0 + 2.0 * ((-1) ** i) for i in range(60)] + other = [100.0 + 1.0 * ((-1) ** i) for i in range(60)] + panel = _flat_panel({"CALM": calm, "WILD": wild, "MED": medium, "OTH": other}) + asof = 1059 + elig = ["CALM", "WILD", "MED", "OTH"] + scores = xs_lowvol_score(panel, elig, asof, window=30) + assert set(scores) == set(elig) + assert scores["CALM"] == max(scores.values()) + assert scores["WILD"] == min(scores.values()) + + +def test_illiquidity_ranks_low_qvol_coins_highest(): + # Same prices, different dollar-volume. Illiquidity longs the THIN coin. + panel: Panel = {} + for sym, qv in {"THIN": 1e5, "FAT": 1e9, "MID": 5e6, "OTH": 2e7}.items(): + panel[sym] = {1000 + i: (100.0, qv, 0.0) for i in range(40)} + asof = 1039 + elig = ["THIN", "FAT", "MID", "OTH"] + scores = xs_illiquidity_score(panel, elig, asof, window=30) + assert set(scores) == set(elig) + assert scores["THIN"] == max(scores.values()) + assert scores["FAT"] == min(scores.values()) + + +def test_scores_aligned_and_robust_to_insufficient_history(): + # NEW has only one bar -> its factor value is None -> robust_z neutralizes to 0.0, + # but it is still present in every returned dict (aligned to `eligible`). + panel = _flat_panel({ + "A": [100.0] * 10 + [100.0, 90.0, 80.0, 70.0], + "B": [100.0] * 10 + [100.0, 110.0, 120.0, 130.0], + "C": [100.0] * 14, + }) + panel["NEW"] = {1013: (100.0, 5e6, 0.0)} # single bar at asof, no history + asof = 1013 + elig = ["A", "B", "C", "NEW"] + for fn in (xs_reversal_score, xs_lowvol_score, xs_illiquidity_score): + scores = fn(panel, elig, asof) + assert set(scores) == set(elig) + assert all(math.isfinite(v) for v in scores.values()) + assert scores["NEW"] == 0.0 # insufficient data -> neutral + + +def test_reversal_is_point_in_time(): + # A coin's FUTURE prices must not change its score at an earlier asof. + base = { + "A": [100.0] * 10 + [100.0, 90.0, 80.0, 70.0], + "B": [100.0] * 10 + [100.0, 110.0, 120.0, 130.0], + "C": [100.0] * 14, + "D": [100.0] * 10 + [100.0, 105.0, 95.0, 100.0], + } + panel = _flat_panel(base) + asof = 1013 + elig = ["A", "B", "C", "D"] + before = xs_reversal_score(panel, elig, asof, lookback=3) + # Append a wild FUTURE move to A after the asof day — must not change the asof score. + panel["A"][1014] = (1000.0, 5e6, 0.0) + panel["A"][1015] = (5.0, 5e6, 0.0) + after = xs_reversal_score(panel, elig, asof, lookback=3) + assert before == after + + +def test_lowvol_is_point_in_time(): + calm = [100.0 + 0.05 * ((-1) ** i) for i in range(60)] + wild = [100.0 + 10.0 * ((-1) ** i) for i in range(60)] + med = [100.0 + 2.0 * ((-1) ** i) for i in range(60)] + oth = [100.0 + 1.0 * ((-1) ** i) for i in range(60)] + panel = _flat_panel({"CALM": calm, "WILD": wild, "MED": med, "OTH": oth}) + asof = 1040 + elig = ["CALM", "WILD", "MED", "OTH"] + before = xs_lowvol_score(panel, elig, asof, window=30) + # future explosion in CALM after asof must not affect its asof score + panel["CALM"][1041] = (10000.0, 5e6, 0.0) + panel["CALM"][1042] = (1.0, 5e6, 0.0) + after = xs_lowvol_score(panel, elig, asof, window=30) + assert before == after