ea6a94f38a
The long-only crypto_tstrend sleeve ignored the perp funding a long book pays (~4.1%/yr drag in a positive-funding bull), overstating the deploy book. TrendRunner now takes an optional funding panel and nets Σ position·funding[nxt] per booked day; weights_and_contributions() nets it per-symbol too so the reconciliation invariant (Σ weight·return == sleeve return) still holds — no live-book/returns-book drift. Both callers (bybit_book_eval._tstrend_series, bybit_edges_eval._tstrend_metrics) and the paper-weights reconciliation now pass store.crypto_funding_panel(). Turnover stays a separate downstream haircut; funding is per-holding so it belongs in the per-symbol contribution. Book Sharpe 1.93 -> 1.85 (honest); long-only still beats long-short (1.49). 1807 tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
151 lines
6.7 KiB
Python
151 lines
6.7 KiB
Python
import json
|
|
|
|
import numpy as np
|
|
|
|
from fxhnt.application.trend_runner import (
|
|
TrendRunner,
|
|
TrendRunResult,
|
|
evaluate_trend,
|
|
)
|
|
|
|
_PPY = 252
|
|
|
|
|
|
def _uptrend_panel(n: int = 300) -> dict[str, dict[int, float]]:
|
|
"""Three assets: one sustained uptrend, two flat-ish (small noise)."""
|
|
rng = np.random.default_rng(0)
|
|
panel: dict[str, dict[int, float]] = {}
|
|
# sustained uptrend
|
|
panel["UP"] = {i: float(100.0 * (1.0 + 0.003) ** i) for i in range(n)}
|
|
# flat with tiny deterministic wobble
|
|
panel["FLAT1"] = {i: float(100.0 + 0.5 * np.sin(i / 7.0)) for i in range(n)}
|
|
panel["FLAT2"] = {i: float(50.0 + 0.3 * np.cos(i / 5.0)) for i in range(n)}
|
|
return panel
|
|
|
|
|
|
def _downtrend_panel(n: int = 300) -> dict[str, dict[int, float]]:
|
|
"""One sustained downtrend asset (for crisis-alpha short test)."""
|
|
return {"DOWN": {i: float(100.0 * (1.0 - 0.003) ** i) for i in range(n)}}
|
|
|
|
|
|
def test_runner_returns_result_shape():
|
|
res = TrendRunner(_uptrend_panel(), min_history=130, vol_window=30).run()
|
|
assert isinstance(res, TrendRunResult)
|
|
assert isinstance(res.returns, np.ndarray)
|
|
assert res.n_rebalances == len(res.returns)
|
|
assert res.n_rebalances > 0
|
|
assert np.isfinite(res.returns).all()
|
|
assert res.n_assets_avg > 0.0
|
|
|
|
|
|
def test_runner_emits_dates_parallel_to_returns():
|
|
res = TrendRunner(_uptrend_panel(), min_history=130, vol_window=30).run()
|
|
assert len(res.dates) == len(res.returns)
|
|
assert res.dates == sorted(res.dates)
|
|
assert len(set(res.dates)) == len(res.dates)
|
|
assert all(isinstance(d, int) for d in res.dates)
|
|
|
|
|
|
def test_captures_uptrend_long():
|
|
res = TrendRunner(_uptrend_panel(), min_history=130, vol_window=30, target_vol=0.10).run()
|
|
# going long the persistent uptrend earns positive cumulative return
|
|
assert float(np.sum(res.returns)) > 0.0
|
|
|
|
|
|
def test_shorts_downtrend_when_long_short():
|
|
ls = TrendRunner(_downtrend_panel(), min_history=130, vol_window=30,
|
|
long_short=True).run()
|
|
lf = TrendRunner(_downtrend_panel(), min_history=130, vol_window=30,
|
|
long_short=False).run()
|
|
# long_short shorts the downtrend -> net positive; long/flat earns ~0 (flat at 0)
|
|
assert float(np.sum(ls.returns)) > 0.0
|
|
assert float(np.sum(ls.returns)) > float(np.sum(lf.returns))
|
|
assert abs(float(np.sum(lf.returns))) < 1e-6 # long-only on a downtrend = stays flat
|
|
|
|
|
|
def test_determinism():
|
|
r1 = TrendRunner(_uptrend_panel(), min_history=130, vol_window=30).run().returns
|
|
r2 = TrendRunner(_uptrend_panel(), min_history=130, vol_window=30).run().returns
|
|
assert np.array_equal(r1, r2)
|
|
|
|
|
|
def test_gross_cap_limits_leverage():
|
|
res = TrendRunner(_uptrend_panel(), min_history=130, vol_window=30,
|
|
target_vol=5.0, gross_cap=1.0).run()
|
|
# even with an absurd target vol, gross is capped -> returns stay bounded per period
|
|
assert np.isfinite(res.returns).all()
|
|
assert np.max(np.abs(res.returns)) < 1.0
|
|
|
|
|
|
def test_funding_overlay_subtracts_position_times_funding():
|
|
"""With a funding panel, each booked day's return is the price-only return MINUS Σ position·funding
|
|
charged on the realized day — the exact perp funding P&L a long book pays. Longs (w>0) at positive
|
|
funding pay, so the net return is strictly lower."""
|
|
panel = _uptrend_panel()
|
|
r = TrendRunner(panel, min_history=130, vol_window=30, long_short=False)
|
|
price = r.run()
|
|
F = 0.001 # constant +10 bp/day funding on every symbol
|
|
funding = {sym: {d: F for d in series} for sym, series in panel.items()}
|
|
net = TrendRunner(panel, min_history=130, vol_window=30, long_short=False, funding=funding).run()
|
|
wc = r.weights_and_contributions() # {realized_day: {sym: (weight, ret)}}
|
|
assert net.dates == price.dates
|
|
for i, day in enumerate(net.dates):
|
|
expected_fc = sum(w * F for (w, _r) in wc[day].values())
|
|
assert abs((price.returns[i] - net.returns[i]) - expected_fc) < 1e-12
|
|
# a long-biased book at positive funding is a NET DRAG
|
|
assert float(np.sum(net.returns)) < float(np.sum(price.returns))
|
|
|
|
|
|
def test_weights_and_contributions_reconcile_with_funding():
|
|
"""The reconciliation invariant survives funding: Σ weight·contribution == run().returns even when a
|
|
funding panel is set. The per-symbol contribution nets the funding each position pays, so the surfaced
|
|
attribution still reproduces the booked (cost-free) return — no live-book/returns-book drift."""
|
|
panel = _uptrend_panel()
|
|
F = 0.001
|
|
funding = {sym: {d: F for d in series} for sym, series in panel.items()}
|
|
# cost_bps=0: the weights reproduce the COST-FREE return (turnover is a separate downstream haircut,
|
|
# by design); with that isolated, Σ weight·contribution must equal run()'s funding-net return.
|
|
r = TrendRunner(panel, min_history=130, vol_window=30, long_short=False, cost_bps=0.0, funding=funding)
|
|
res = r.run()
|
|
wc = r.weights_and_contributions()
|
|
for i, day in enumerate(res.dates):
|
|
implied = sum(w * ret for (w, ret) in wc[day].values())
|
|
assert abs(implied - float(res.returns[i])) < 1e-12
|
|
|
|
|
|
def test_no_funding_panel_leaves_returns_unchanged():
|
|
"""funding=None (default) is exactly the price-only behavior — no silent change to existing tracks."""
|
|
panel = _uptrend_panel()
|
|
a = TrendRunner(panel, min_history=130, vol_window=30).run().returns
|
|
b = TrendRunner(panel, min_history=130, vol_window=30, funding=None).run().returns
|
|
assert np.array_equal(a, b)
|
|
|
|
|
|
def test_short_positions_receive_funding_credit():
|
|
"""Sign correctness: a SHORT (w<0) at positive funding RECEIVES funding, so long_short net return
|
|
EXCEEDS the price-only return on a shorted downtrend."""
|
|
panel = _downtrend_panel()
|
|
F = 0.001
|
|
funding = {sym: {d: F for d in series} for sym, series in panel.items()}
|
|
price = TrendRunner(panel, min_history=130, vol_window=30, long_short=True).run()
|
|
net = TrendRunner(panel, min_history=130, vol_window=30, long_short=True, funding=funding).run()
|
|
assert float(np.sum(net.returns)) > float(np.sum(price.returns))
|
|
|
|
|
|
def test_evaluate_trend_matrix():
|
|
panel = _uptrend_panel()
|
|
configs = [
|
|
{"label": "fast", "lookbacks": (10, 20, 40)},
|
|
{"label": "slow", "lookbacks": (20, 60, 120)},
|
|
{"label": "long_only", "lookbacks": (20, 60, 120), "long_short": False},
|
|
]
|
|
out = evaluate_trend(panel, periods_per_year=_PPY, configs=configs)
|
|
assert out["n_trials"] == len(configs)
|
|
assert set(out["cells"]) == {"fast", "slow", "long_only"}
|
|
for label, cell in out["cells"].items():
|
|
assert cell["config_label"] == label
|
|
assert "stats" in cell and "verdict" in cell
|
|
assert cell["verdict"]["n_trials"] == len(configs)
|
|
# JSON-serializable (NaN-safe)
|
|
json.dumps(out)
|