182 lines
8.7 KiB
Python
182 lines
8.7 KiB
Python
"""Web smoke for the Bybit 4-edge book in the cockpit sim: the book selector loads the precomputed
|
|
`bybit_sleeve_ret` table, runs the NAIVE equal-weight sim with the cheap capital/start/cost knobs, and
|
|
renders the configurable BACKTEST clearly distinguished from the LIVE FORWARD track — WITHOUT ever
|
|
recomputing the heavy 4-edge book on the request path.
|
|
|
|
Asserts:
|
|
* /paper/sim?book=bybit_4edge renders the book selector + a curve labelled "Backtest (configurable)";
|
|
* /paper/sim/run?book=bybit_4edge returns a curve + metrics computed from bybit_sleeve_ret;
|
|
* capital scales end-value; cost_bps changes the return (the cheap knobs work);
|
|
* the request path NEVER calls the heavy edge computation (a tripwire on sleeve_returns_from_store);
|
|
* the Bybit view shows the live forward track (labelled distinctly) with its gate status / days;
|
|
* the existing Binance sim still works and defaults to binance_combined.
|
|
NO network — in-memory SQLite repos, seeded directly.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import json
|
|
import re
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
|
|
from fxhnt.adapters.persistence.paper_repo import PaperRepo
|
|
from fxhnt.adapters.web.app import create_app
|
|
from fxhnt.application.forward_models import ForwardNavRow as NavRowDTO
|
|
from fxhnt.application.forward_models import ForwardSummary
|
|
|
|
_BYBIT_SLEEVES = {"crypto_tstrend": 0.004, "unlock": 0.0025, "xsfunding": 0.001, "positioning": 0.0018}
|
|
|
|
|
|
def _seeded_paper_repo() -> PaperRepo:
|
|
"""A repo with ~120 days of precomputed BYBIT per-sleeve returns in bybit_sleeve_ret."""
|
|
repo = PaperRepo("sqlite://")
|
|
repo.migrate()
|
|
at = dt.datetime(2026, 6, 24, tzinfo=dt.UTC)
|
|
base = dt.date(2026, 1, 1)
|
|
for i in range(120):
|
|
d = (base + dt.timedelta(days=i)).isoformat()
|
|
for sleeve, amp in _BYBIT_SLEEVES.items():
|
|
ret = amp * (1.0 if (i + hash(sleeve)) % 3 else -0.8)
|
|
repo.upsert_bybit_sleeve_ret(d, sleeve, ret, at=at)
|
|
return repo
|
|
|
|
|
|
def _seeded_forward_repo(*, with_track: bool = True) -> ForwardNavRepo:
|
|
fwd = ForwardNavRepo("sqlite://")
|
|
fwd.migrate()
|
|
if with_track:
|
|
at = dt.datetime(2026, 6, 24)
|
|
base = dt.date(2026, 4, 1)
|
|
nav = 1.0
|
|
rows = []
|
|
for i in range(15):
|
|
nav *= 1.001
|
|
rows.append(NavRowDTO(strategy_id="bybit_4edge",
|
|
date=(base + dt.timedelta(days=i)).isoformat(), ret=0.001, nav=nav))
|
|
fwd.upsert_rows(rows, at=at)
|
|
sm = ForwardSummary(strategy_id="bybit_4edge", as_of="2026-04-15", days=15, nav=nav,
|
|
total_return=nav - 1.0, sharpe=1.2, maxdd=-0.02)
|
|
fwd.upsert_summary(sm, gate_status="WAIT", gate_reason="15/60 forward days", at=at)
|
|
return fwd
|
|
|
|
|
|
def _client(paper: PaperRepo, fwd: ForwardNavRepo | None = None) -> TestClient:
|
|
return TestClient(create_app(fwd or _seeded_forward_repo(), paper_repo=paper))
|
|
|
|
|
|
def _sim_data(html: str) -> dict:
|
|
m = re.search(r'<script type="application/json" id="sim-data">(.*?)</script>', html, re.DOTALL)
|
|
assert m, "embedded #sim-data JSON not found"
|
|
return json.loads(m.group(1))
|
|
|
|
|
|
def test_bybit_page_renders_selector_and_lazy_loads_backtest() -> None:
|
|
c = _client(_seeded_paper_repo())
|
|
r = c.get("/paper/sim", params={"book": "bybit_4edge"})
|
|
assert r.status_code == 200
|
|
assert "Bybit 4-edge" in r.text # book selector (painted instantly)
|
|
assert "naive equal-weight" in r.text # naive eq-wt note (form-side, not the overlay)
|
|
# The backtest curve/label is lazy-loaded for the bybit book too (hx-get carries book=bybit_4edge).
|
|
assert 'hx-get="/paper/sim/run' in r.text and "book=bybit_4edge" in r.text
|
|
assert "<svg" not in r.text
|
|
|
|
|
|
def test_bybit_run_fragment_has_backtest_label_and_curve() -> None:
|
|
c = _client(_seeded_paper_repo())
|
|
r = c.get("/paper/sim/run",
|
|
params={"book": "bybit_4edge",
|
|
"sleeves": "crypto_tstrend,unlock,xsfunding,positioning"})
|
|
assert r.status_code == 200
|
|
assert "Backtest (configurable)" in r.text # the backtest is labelled
|
|
assert "<svg" in r.text
|
|
|
|
|
|
def test_bybit_run_returns_curve_and_metrics_from_precomputed_table() -> None:
|
|
c = _client(_seeded_paper_repo())
|
|
r = c.get("/paper/sim/run",
|
|
params={"book": "bybit_4edge", "capital": 50000,
|
|
"sleeves": "crypto_tstrend,unlock,xsfunding,positioning"})
|
|
assert r.status_code == 200
|
|
assert "<svg" in r.text and "CAGR" in r.text and "Sharpe" in r.text
|
|
data = _sim_data(r.text)
|
|
assert data["dates"] and data["equity"]
|
|
assert 40000 < data["equity"][0] < 60000 # curve starts near the configured capital
|
|
|
|
|
|
def test_bybit_capital_and_cost_are_cheap_knobs() -> None:
|
|
c = _client(_seeded_paper_repo())
|
|
sl = "crypto_tstrend,unlock,xsfunding,positioning"
|
|
small = c.get("/paper/sim/run", params={"book": "bybit_4edge", "capital": 50000, "sleeves": sl})
|
|
big = c.get("/paper/sim/run", params={"book": "bybit_4edge", "capital": 200000, "sleeves": sl})
|
|
# capital scales end-value ~4x (linear).
|
|
e_small = _sim_data(small.text)["equity"][-1]
|
|
e_big = _sim_data(big.text)["equity"][-1]
|
|
assert abs(e_big / e_small - 4.0) < 0.01
|
|
# cost_bps changes the return.
|
|
cheap = c.get("/paper/sim/run", params={"book": "bybit_4edge", "capital": 100000, "sleeves": sl,
|
|
"cost_bps": 0})
|
|
pricey = c.get("/paper/sim/run", params={"book": "bybit_4edge", "capital": 100000, "sleeves": sl,
|
|
"cost_bps": 20})
|
|
assert _sim_data(cheap.text)["equity"][-1] != _sim_data(pricey.text)["equity"][-1]
|
|
|
|
|
|
def test_bybit_run_does_not_recompute_the_heavy_edges(monkeypatch) -> None:
|
|
"""Tripwire: the sim request path reads bybit_sleeve_ret and must NEVER call the heavy 4-edge runner
|
|
(`sleeve_returns_from_store`). Detonate it and prove the sim still serves a curve."""
|
|
import fxhnt.application.bybit_book_eval as bbe
|
|
|
|
def _boom(*a, **k):
|
|
raise AssertionError("heavy 4-edge compute called on a web request — must read bybit_sleeve_ret!")
|
|
|
|
monkeypatch.setattr(bbe, "sleeve_returns_from_store", _boom)
|
|
c = _client(_seeded_paper_repo())
|
|
r = c.get("/paper/sim/run",
|
|
params={"book": "bybit_4edge", "capital": 100000,
|
|
"sleeves": "crypto_tstrend,unlock,xsfunding,positioning"})
|
|
assert r.status_code == 200
|
|
assert _sim_data(r.text)["dates"] # got a real curve, tripwire never fired
|
|
|
|
|
|
def test_bybit_view_shows_live_forward_track_with_gate() -> None:
|
|
# The forward track renders in the lazy-loaded result fragment (same as the backtest curve).
|
|
c = _client(_seeded_paper_repo(), _seeded_forward_repo(with_track=True))
|
|
r = c.get("/paper/sim/run", params={"book": "bybit_4edge",
|
|
"sleeves": "crypto_tstrend,unlock,xsfunding,positioning"})
|
|
assert r.status_code == 200
|
|
assert "Live forward" in r.text # forward track labelled distinctly
|
|
assert "Backtest (configurable)" in r.text # both present on the same fragment
|
|
assert "15/60 forward days" in r.text # gate status / days
|
|
assert "WAIT" in r.text
|
|
|
|
|
|
def test_bybit_view_handles_no_forward_data() -> None:
|
|
fwd = ForwardNavRepo("sqlite://")
|
|
fwd.migrate() # registry seeds bybit_4edge but no nav rows
|
|
c = _client(_seeded_paper_repo(), fwd)
|
|
r = c.get("/paper/sim/run", params={"book": "bybit_4edge",
|
|
"sleeves": "crypto_tstrend,unlock,xsfunding,positioning"})
|
|
assert r.status_code == 200
|
|
assert "Live forward" in r.text
|
|
assert "no live forward NAV recorded yet" in r.text
|
|
|
|
|
|
def test_binance_sim_still_default_and_works() -> None:
|
|
"""The existing Binance sim is unchanged: default book is binance_combined, reads paper_sleeve_ret."""
|
|
repo = PaperRepo("sqlite://")
|
|
repo.migrate()
|
|
at = dt.datetime(2026, 6, 24, tzinfo=dt.UTC)
|
|
base = dt.date(2026, 1, 1)
|
|
for i in range(120):
|
|
d = (base + dt.timedelta(days=i)).isoformat()
|
|
for sleeve in ("crypto_tstrend", "unlock"):
|
|
repo.upsert_sleeve_ret(d, sleeve, 0.003 * (1.0 if i % 2 else -0.8), at=at)
|
|
c = _client(repo)
|
|
r = c.get("/paper/sim") # no book param → binance_combined
|
|
assert r.status_code == 200
|
|
assert "Backtest (configurable)" not in r.text # the bybit-only label is absent
|
|
run = c.get("/paper/sim/run", params={"capital": 100000, "sleeves": "crypto_tstrend,unlock"})
|
|
assert run.status_code == 200
|
|
assert _sim_data(run.text)["dates"]
|