123 lines
4.8 KiB
Python
123 lines
4.8 KiB
Python
"""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)
|