fd2316b1e8
Verifying the Backtest page: the scrub state panel labelled n_active "active sleeves", but n_active counts the instruments HELD that day — the Bybit book's ~64-coin liquid universe (compare_measured_cost sets n_active=[n_coins]*len), the ETF book's ~5 ETFs — NOT the book's sleeves (its 4 edges). So a 4-edge book read as having "64 sleeves". Relabelled "active positions", accurate for both the coin book and the ETF book. Regression test pins "active positions" (and asserts "active sleeves" absent). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
231 lines
11 KiB
Python
231 lines
11 KiB
Python
"""Web smoke for the Bybit 4-edge book in the cockpit Backtest: the book selector loads the precomputed
|
|
per-coin MEASURED-cost curve (the honest real-cost artifact) and renders it scaled by capital / windowed by
|
|
period, clearly distinguished from the LIVE FORWARD track — WITHOUT ever recomputing the heavy 4-edge book on
|
|
the request path. There is NO configurable-cost sandbox (cost is a measured fact, not a slider).
|
|
|
|
Asserts:
|
|
* /paper/sim?book=bybit_4edge renders the book selector + lazy-loads the result;
|
|
* /paper/sim/run?book=bybit_4edge returns the measured curve + metrics read from the precomputed artifact;
|
|
* capital scales end-value (the curve rescales linearly);
|
|
* 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;
|
|
* bybit_4edge is the default book, and a stale binance_combined book param (Task 7a removed the Binance
|
|
venue) degrades gracefully to it — never a 500.
|
|
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.compare_measured_cost import SINGLE_CACHE_KEYS
|
|
from fxhnt.application.forward_models import ForwardNavRow as NavRowDTO
|
|
from fxhnt.application.forward_models import ForwardSummary
|
|
from fxhnt.application.paper_sim import _metrics
|
|
|
|
_BYBIT_SLEEVES = {"crypto_tstrend": 0.004, "unlock": 0.0025, "xsfunding": 0.001, "positioning": 0.0018}
|
|
|
|
|
|
def _seed_measured_cache(repo: PaperRepo, book: str = "bybit_4edge", *,
|
|
start: dt.date = dt.date(2022, 1, 1), days: int = 200,
|
|
capital: float = 100_000.0) -> None:
|
|
"""Persist a synthetic single-book MEASURED-cost payload — the honest real-cost curve the Backtest reads.
|
|
The cockpit reads ONLY this precomputed artifact on the request path (no heavy per-coin compute)."""
|
|
iso, equity, drawdown, net_rets, epoch = [], [], [], [], []
|
|
eq, peak = capital, capital
|
|
for i in range(days):
|
|
d = start + dt.timedelta(days=i)
|
|
r = 0.004 * (1.0 if (i + hash(book)) % 3 else -0.8)
|
|
eq *= (1.0 + r)
|
|
peak = max(peak, eq)
|
|
iso.append(d.isoformat()); equity.append(eq); net_rets.append(r)
|
|
drawdown.append((eq - peak) / peak if peak > 0 else 0.0)
|
|
epoch.append((d - dt.date(1970, 1, 1)).days)
|
|
payload = {
|
|
"book": book, "dates_iso": iso, "equity": equity, "drawdown": drawdown,
|
|
"leverage": [1.0] * days, "killed": [False] * days, "n_active": [4] * days,
|
|
"metrics": _metrics(epoch, equity, net_rets, drawdown, capital),
|
|
"median_spread_bps": 1.0, "cost_drag_bps": 6.5,
|
|
"used_real_spread": book == "bybit_4edge", "capital": capital, "taker_fee_bps": 5.5,
|
|
}
|
|
repo.upsert_compare_measured_cache(SINGLE_CACHE_KEYS[book],
|
|
payload, at=dt.datetime(2026, 6, 24, tzinfo=dt.UTC))
|
|
|
|
|
|
def _seeded_paper_repo(*, with_cache: bool = True) -> PaperRepo:
|
|
"""A repo with ~120 days of precomputed BYBIT per-sleeve returns in bybit_sleeve_ret PLUS (by default) the
|
|
precomputed measured-cost artifact the Backtest renders."""
|
|
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)
|
|
if with_cache:
|
|
_seed_measured_cache(repo, "bybit_4edge")
|
|
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)
|
|
# The curve 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", "capital": 100000})
|
|
assert r.status_code == 200
|
|
assert "Backtest (real measured cost)" in r.text # the honest real-cost backtest is labelled
|
|
assert 'data-cost-mode="measured"' in r.text # the measured-cost marker
|
|
assert "<svg" in r.text
|
|
# The scrub state-panel counts the instruments/coins HELD, not the book's sleeves — a 4-edge book holds
|
|
# ~64 coins, so it must read "active positions", never "active sleeves" (which read as "64 sleeves").
|
|
assert "active positions" in r.text and "active sleeves" not in r.text
|
|
|
|
|
|
def _visible_text(html: str) -> str:
|
|
"""The user-FACING text of a rendered page: `<style>`/`<script>` blocks dropped and every HTML tag
|
|
(with its attributes) stripped, so a `data-cost-drag="6.50"` attribute or a `title="..."` string can't
|
|
false-trip a plain-language copy assertion — only what a reader actually sees is left."""
|
|
html = re.sub(r"<(style|script)\b[^>]*>.*?</\1>", " ", html, flags=re.DOTALL | re.IGNORECASE)
|
|
return re.sub(r"<[^>]+>", " ", html).lower()
|
|
|
|
|
|
def test_bybit_measured_cost_caption_shows_plain_percent_not_bp() -> None:
|
|
# MINOR whole-branch-review fix: "avg drag X bp" leaked visible `bp` jargon. The seeded artifact's
|
|
# cost_drag_bps=6.5 -> 0.07% (6.5 / 100, 2dp) is now the rendered avg-drag figure, plain-language.
|
|
c = _client(_seeded_paper_repo())
|
|
r = c.get("/paper/sim/run", params={"book": "bybit_4edge", "capital": 100000})
|
|
assert r.status_code == 200
|
|
assert "avg drag 0.07%" in r.text
|
|
text = _visible_text(r.text)
|
|
assert "bp" not in text and "bps" not in text
|
|
|
|
|
|
def test_bybit_run_returns_curve_and_metrics_from_precomputed_artifact() -> None:
|
|
c = _client(_seeded_paper_repo())
|
|
r = c.get("/paper/sim/run", params={"book": "bybit_4edge", "capital": 50000})
|
|
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_scales_the_measured_curve() -> None:
|
|
c = _client(_seeded_paper_repo())
|
|
small = c.get("/paper/sim/run", params={"book": "bybit_4edge", "capital": 50000})
|
|
big = c.get("/paper/sim/run", params={"book": "bybit_4edge", "capital": 200000})
|
|
# capital scales end-value ~4x (linear) — the only cost is the measured fact baked into the curve.
|
|
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
|
|
|
|
|
|
def test_bybit_run_does_not_recompute_the_heavy_edges(monkeypatch) -> None:
|
|
"""Tripwire: the Backtest request path reads the precomputed measured artifact and must NEVER call the
|
|
heavy 4-edge runner (`sleeve_returns_from_store`). Detonate it and prove the curve still serves."""
|
|
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 the cached artifact!")
|
|
|
|
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})
|
|
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 (alongside the backtest curve).
|
|
c = _client(_seeded_paper_repo(), _seeded_forward_repo(with_track=True))
|
|
r = c.get("/paper/sim/run", params={"book": "bybit_4edge", "capital": 100000})
|
|
assert r.status_code == 200
|
|
assert "Live forward" in r.text # forward track labelled distinctly
|
|
assert "Backtest (real measured cost)" 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", "capital": 100000})
|
|
assert r.status_code == 200
|
|
assert "Live forward" in r.text
|
|
assert "no live forward NAV recorded yet" in r.text
|
|
|
|
|
|
def test_bybit_is_the_default_book() -> None:
|
|
"""The default book is the Bybit 4-edge DEPLOY book. A bare /paper/sim pins bybit in the form + the
|
|
lazy-load result URL."""
|
|
c = _client(_seeded_paper_repo())
|
|
r = c.get("/paper/sim") # no book param → bybit_4edge (the deploy fund)
|
|
assert r.status_code == 200
|
|
assert 'data-book="bybit_4edge"' in r.text
|
|
m = re.search(r'hx-get="(/paper/sim/run\?[^"]+)"', r.text)
|
|
assert m, "lazy-load result URL not found"
|
|
assert "book=bybit_4edge" in m.group(1)
|
|
|
|
|
|
def test_stale_binance_book_param_degrades_to_the_bybit_default() -> None:
|
|
"""Task 7a removed the Binance venue + book from the cockpit. An old bookmarked
|
|
/paper/sim?book=binance_combined URL must degrade gracefully to the bybit default — never 500, never
|
|
render a Binance-labelled curve."""
|
|
repo = PaperRepo("sqlite://")
|
|
repo.migrate()
|
|
_seed_measured_cache(repo, "bybit_4edge")
|
|
c = _client(repo)
|
|
r = c.get("/paper/sim", params={"book": "binance_combined"})
|
|
assert r.status_code == 200
|
|
assert 'data-book="bybit_4edge"' in r.text
|
|
run = c.get("/paper/sim/run", params={"book": "binance_combined", "capital": 100000})
|
|
assert run.status_code == 200
|
|
assert "Backtest (real measured cost)" in run.text # rendered the bybit book, not a 404/500
|
|
assert _sim_data(run.text)["dates"]
|