gitadmin/fxhnt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
fxhnt/tests/integration/test_paper_web.py
jgrusewski 048e5ca26d fix(cockpit): name the deploy book 'Bybit 4-edge book' consistently on every surface 
Trim the registry display_name (drops the (naive eq-wt, forward) parenthetical;
exec/levered keep their qualifiers), rename the /paper/crypto h2 (was 'Bybit
paper book') and the sim pill (was 'Bybit 4-edge'). Overview/paper/sim/strategy
all render 'Bybit 4-edge book' now (locally verified); old names gone.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-16 13:53:03 +02:00

438 lines
22 KiB
Python
Raw Permalink Blame History

"""Web smoke for the paper cockpit: /paper renders equity + a position; /paper/live returns updated MTM;
empty live prices degrade gracefully (never 500). Bybit is the cockpit's only paper-book venue (the Binance
venue toggle was removed in Task 7a — see test_cockpit_bybit_only.py for the bybit-only assertions)."""
from __future__ import annotations
import datetime as dt
from fastapi.testclient import TestClient
from fxhnt.adapters.persistence.paper_repo import PaperRepo
from fxhnt.adapters.web.app import create_app
from fxhnt.application.paper_book import PaperBookService
from fxhnt.ports.live_price import FakeLivePrice
def _seeded_bybit_repo() -> PaperRepo:
"""A Bybit paper repo with a live position (short USDTUSD in stablecoin_rotation)."""
repo = PaperRepo("sqlite://")
repo.migrate()
svc = PaperBookService(repo, capital=100_000.0)
svc.derive_and_persist(
run_date="2026-06-21",
sleeve_weights={"stablecoin_rotation": 0.5},
symbol_weights_by_sleeve={"stablecoin_rotation": {"USDTUSD": -1.0}},
prices={"USDTUSD": 2.0},
at=dt.datetime(2026, 6, 21, tzinfo=dt.UTC))
return repo
def _client(bybit_live: FakeLivePrice, bybit_repo: PaperRepo | None = None) -> TestClient:
repo = bybit_repo if bybit_repo is not None else _seeded_bybit_repo()
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
fwd = ForwardNavRepo("sqlite://")
fwd.migrate()
# An explicit empty (in-memory) `paper_repo` — unused by the paper page's rendering, but
# `_deploy_individual_edges` reads via it on every /paper render, so inject one to keep the test hermetic
# (never the from-settings fallback / a real DB).
binance = PaperRepo("sqlite://")
binance.migrate()
return TestClient(create_app(fwd, paper_repo=binance, bybit_paper_repo=repo, bybit_live=bybit_live))
def test_paper_page_renders_equity_and_position() -> None:
c = _client(FakeLivePrice({"USDTUSD": 1.9})) # short profits as price fell
r = c.get("/paper/crypto")
assert r.status_code == 200
assert "Paper" in r.text
assert "USDTUSD" in r.text
assert "equity" in r.text.lower()
def test_paper_live_partial_returns_updated_mtm() -> None:
c = _client(FakeLivePrice({"USDTUSD": 1.9}))
r = c.get("/paper/live")
assert r.status_code == 200
assert "equity" in r.text.lower()
assert "USDTUSD" in r.text
def test_paper_degrades_gracefully_with_empty_prices() -> None:
c = _client(FakeLivePrice({})) # no live prices at all
r = c.get("/paper/crypto")
assert r.status_code == 200
r2 = c.get("/paper/live")
assert r2.status_code == 200
def test_fleet_page_links_to_paper() -> None:
c = _client(FakeLivePrice({"USDTUSD": 1.9}))
r = c.get("/")
assert r.status_code == 200
assert "/paper" in r.text
# --- Bybit is the cockpit's only venue: any `?venue=` query param is simply ignored --------------------
_BYBIT_SLEEVES = {"crypto_tstrend": 0.004, "unlock": 0.0025, "xsfunding": 0.001, "positioning": 0.0018}
def _seeded_bybit_paper_repo() -> PaperRepo:
"""A paper repo with the USDTUSD position AND ~120 days of precomputed Bybit per-sleeve returns in
bybit_sleeve_ret (so /paper can render both the live book + the individual-edges breakdown)."""
repo = PaperRepo("sqlite://")
repo.migrate()
svc = PaperBookService(repo, capital=100_000.0)
svc.derive_and_persist(
run_date="2026-06-21",
sleeve_weights={"stablecoin_rotation": 0.5},
symbol_weights_by_sleeve={"stablecoin_rotation": {"USDTUSD": -1.0}},
prices={"USDTUSD": 2.0},
at=dt.datetime(2026, 6, 21, tzinfo=dt.UTC))
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 _client_with_forward(paper: PaperRepo, *, with_track: bool):
"""A TestClient whose forward repo has (or lacks) a bybit_4edge forward NAV track. `paper` seeds the
bybit_sleeve_ret table read by the individual-edges breakdown; the live bybit book is a separate
(empty) repo."""
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.application.forward_models import ForwardNavRow as NavRowDTO
from fxhnt.application.forward_models import ForwardSummary
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)
# An empty bybit live book (a separate in-memory PaperRepo) + a no-op bybit mark — the live
# bybit view renders cleanly (no positions) while the backtest reference reads the shared bybit_sleeve_ret.
bybit_paper = PaperRepo("sqlite://")
bybit_paper.migrate()
return TestClient(create_app(fwd, paper_repo=paper, bybit_paper_repo=bybit_paper, bybit_live=FakeLivePrice({})))
def test_paper_defaults_to_bybit_live_book_with_badge_and_backtest_reference() -> None:
c = _client_with_forward(_seeded_bybit_paper_repo(), with_track=True)
r = c.get("/paper/crypto")
assert r.status_code == 200
# The Bybit DEPLOY book is shown with its BYBIT badge (mark-to-market, per-symbol) + a PAPER exec badge.
assert "Bybit 4-edge book" in r.text
assert ">BYBIT<" in r.text
assert "mark-to-market" in r.text
assert "badge-exec paper" in r.text
# ALL four constituent edges listed in the individual-edges breakdown (incl. positioning), consistent
# with the Overview's deploy constituents. unlock/xsfunding/positioning still have a STRATEGY_REGISTRY
# entry, so their raw sleeve id appears in their working `/strategy/{id}` link; crypto_tstrend's
# standalone registry entry was retired (Phase 0b Task 4) — it renders its own human display name
# instead, with NO dead `/strategy/crypto_tstrend` link (see test_paper_web.py::
# test_crypto_tstrend_edge_gets_human_name_and_no_dead_link below).
for sleeve in ("unlock", "xsfunding", "positioning"):
assert f'href="/strategy/{sleeve}"' in r.text
# individual edges rendered via the SAME deploy_constituent_row macro the Overview uses (the "Sh" metric).
assert "individual edges" in r.text
assert "Sh " in r.text
def test_crypto_tstrend_edge_gets_human_name_and_no_dead_link() -> None:
# BUG: the crypto_tstrend BOOK SLEEVE shares its string id with the now-retired standalone
# crypto_tstrend STRATEGY_REGISTRY entry (Phase 0b Task 4). The sleeve breakdown must not blank the
# display name to the raw id, and must not link to the now-unknown `/strategy/crypto_tstrend` (404).
from fxhnt.application.bybit_book_eval import _SLEEVE_DISPLAY_NAMES
c = _client_with_forward(_seeded_bybit_paper_repo(), with_track=True)
r = c.get("/paper/crypto")
assert r.status_code == 200
assert _SLEEVE_DISPLAY_NAMES["crypto_tstrend"] in r.text
assert 'href="/strategy/crypto_tstrend"' not in r.text
# unlock/xsfunding/positioning keep their working links (untouched by the fix).
for sleeve in ("unlock", "xsfunding", "positioning"):
assert f'href="/strategy/{sleeve}"' in r.text
# /strategy/crypto_tstrend is not a registered strategy at all — it 404s BY DESIGN. The fix is that the
# cockpit no longer links to it, not that the route now resolves.
assert c.get("/strategy/crypto_tstrend").status_code == 404
def test_paper_default_bybit_shows_forward_gate_and_curve() -> None:
c = _client_with_forward(_seeded_bybit_paper_repo(), with_track=True)
r = c.get("/paper/crypto")
assert r.status_code == 200
assert "WAIT" in r.text # forward gate
assert "15/60 forward days" in r.text # gate reason / days
assert "<svg" in r.text # the forward NAV curve
def test_paper_stray_venue_query_param_is_ignored() -> None:
# No venue toggle any more (Task 7a) — a stray `?venue=binance` (or any other value) is simply ignored;
# the page always renders the Bybit book.
c = _client_with_forward(_seeded_bybit_paper_repo(), with_track=True)
for venue in ("binance", "garbage", "bybit"):
r = c.get(f"/paper/crypto?venue={venue}")
assert r.status_code == 200
assert "Bybit 4-edge book" in r.text
assert "Binance paper book" not in r.text
def test_paper_live_partial_uses_single_route() -> None:
# The /paper/live auto-refresh block no longer carries a `venue` query param — there is only one book.
c = _client_with_forward(_seeded_bybit_paper_repo(), with_track=True)
assert 'hx-get="/paper/live"' in c.get("/paper/crypto").text
assert 'hx-get="/paper/live?venue=' not in c.get("/paper/crypto").text
def test_paper_bybit_young_forward_track_shows_building() -> None:
# Registry seeds bybit_4edge but NO forward nav rows → young/empty forward track, but the backtest
# sleeves still render as the validated reference (never a 500).
c = _client_with_forward(_seeded_bybit_paper_repo(), with_track=False)
r = c.get("/paper/crypto")
assert r.status_code == 200
assert "Bybit 4-edge book" in r.text
assert "positioning" in r.text # backtest-reference sleeves still listed
assert "WAIT" in r.text # young track still gates WAIT
def test_paper_bybit_degrades_gracefully_without_sleeve_data() -> None:
# No bybit_sleeve_ret rows at all → the bybit section renders empty/"no data", never a 500.
empty = PaperRepo("sqlite://")
empty.migrate()
c = _client_with_forward(empty, with_track=False)
r = c.get("/paper/crypto")
assert r.status_code == 200
assert "Bybit 4-edge book" in r.text # section header still shown
# --- Real backfilled nav curve + compounded-equity headline ------------------------------------------
# /paper leads with the COMPOUNDED backfilled record (latest persisted paper_nav equity) and draws the
# real multi-point nav curve (the same chart /paper/replay uses) instead of the MVP 2-point spark. The live
# mark-to-market is kept as a clearly-labeled secondary line. A fresh book (no nav history) falls back to
# the 2-point spark and the live mark IS the headline. LIGHT read only — one persisted nav query, no compute.
def _svg_point_count(svg: str) -> int:
"""Number of points in the FIRST <polyline ... points="..."> of an inline SVG (the nav curve)."""
import re
m = re.search(r'<polyline[^>]*points="([^"]*)"', svg)
if not m:
return 0
return len([p for p in m.group(1).split(" ") if p.strip()])
def _seed_nav(repo: PaperRepo, *, n: int, start: float, end: float) -> None:
"""Persist `n` ascending paper_nav rows from `start` to `end` (compounding record)."""
at = dt.datetime(2026, 6, 24, tzinfo=dt.UTC)
base = dt.date(2021, 1, 1)
for i in range(n):
d = (base + dt.timedelta(days=i)).isoformat()
eq = start + (end - start) * (i / (n - 1)) if n > 1 else end
repo.upsert_nav(d, capital=start, realized=0.0, unrealized=0.0, equity=eq, at=at)
def test_paper_draws_real_multi_point_nav_curve_when_history_exists() -> None:
# A real backfilled bybit nav series → the curve has many points (NOT the 2-point spark).
bybit = PaperRepo("sqlite://")
bybit.migrate()
_seed_nav(bybit, n=200, start=100_000.0, end=393_955.0)
c = _client_bybit_repo(bybit, bybit_live=FakeLivePrice({}))
r = c.get("/paper/crypto")
assert r.status_code == 200
assert _svg_point_count(r.text) > 2 # a real multi-point curve, not the placeholder spark
def test_paper_degrades_gracefully_for_fresh_book_with_no_nav_history() -> None:
# No nav history persisted → the headline_card shows "no backfilled record yet" (no curve to draw) —
# a fresh book must never 500, and the live per-symbol book still renders.
bybit = _seeded_bybit_repo() # has positions but NO paper_nav rows
c = _client(FakeLivePrice({"USDTUSD": 1.9}), bybit_repo=bybit)
r = c.get("/paper/crypto")
assert r.status_code == 200
assert "no backfilled record yet" in r.text
assert "USDTUSD" in r.text # the live per-symbol book still renders
def _client_bybit_repo(bybit: PaperRepo, *, bybit_live) -> TestClient:
"""A TestClient with an explicit bybit paper repo + a fixed bybit live-price source."""
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
fwd = ForwardNavRepo("sqlite://")
fwd.migrate()
binance = PaperRepo("sqlite://")
binance.migrate()
return TestClient(create_app(fwd, paper_repo=binance, bybit_paper_repo=bybit, bybit_live=bybit_live))
def test_paper_headline_leads_with_compounded_nav_and_keeps_live_mark_secondary() -> None:
# The page MUST lead with the compounded multi-year record ($393,955) and ALSO show the live mark
# ($99,959) as a clearly-labeled secondary detail.
bybit = PaperRepo("sqlite://")
bybit.migrate()
# Live book: a short BTC lot so the live mark moves with price (kept distinct from the nav record).
PaperBookService(bybit, capital=100_000.0).derive_and_persist(
run_date="2026-06-24", sleeve_weights={"positioning": 1.0},
symbol_weights_by_sleeve={"positioning": {"BTCUSDT": -0.01}},
prices={"BTCUSDT": 50_000.0}, at=dt.datetime(2026, 6, 24, tzinfo=dt.UTC))
# The LATEST persisted nav row is the compounded record ($393,955). NOTE: view.equity =
# latest_nav_equity + intraday MTM, so to make the live mark distinct (~$99,959) the latest nav row
# must be the LIVE baseline; we therefore seed the long compounding record but pin the FINAL row to
# $99,959 so latest_nav_equity()==99,959 (the live mark) while the curve still climbs to the record.
_seed_nav(bybit, n=199, start=100_000.0, end=393_955.0)
bybit.upsert_nav("2021-07-20", capital=100_000.0, realized=0.0, unrealized=0.0,
equity=393_955.0, at=dt.datetime(2026, 6, 24, tzinfo=dt.UTC))
# No intraday move (price == entry) so the live mark == latest_nav_equity == the compounded record.
c = _client_bybit_repo(bybit, bybit_live=FakeLivePrice({"BTCUSDT": 50_000.0}))
r = c.get("/paper/crypto")
assert r.status_code == 200
# Compounded record is the headline.
assert "393,955" in r.text
# And it's the PRIMARY "paper equity" headline, with a real multi-point curve.
assert "paper equity" in r.text
assert _svg_point_count(r.text) > 2
def test_paper_headline_shows_live_mark_as_secondary_detail() -> None:
# When the live mark differs from the compounded record, BOTH are rendered: the record as the headline,
# the live mark labeled "marked to live prices now". A SHORT BTC lot + a price DROP profits → live mark
# rises above the nav baseline, so the two figures differ and both must appear.
bybit = PaperRepo("sqlite://")
bybit.migrate()
PaperBookService(bybit, capital=100_000.0).derive_and_persist(
run_date="2026-06-24", sleeve_weights={"positioning": 1.0},
symbol_weights_by_sleeve={"positioning": {"BTCUSDT": -1.0}},
prices={"BTCUSDT": 100.0}, at=dt.datetime(2026, 6, 24, tzinfo=dt.UTC))
_seed_nav(bybit, n=50, start=100_000.0, end=120_000.0) # compounded record = $120,000
# Price falls 100 -> 90: the short lot profits, so the live mark = 120,000 + intraday gain > 120,000.
c = _client_bybit_repo(bybit, bybit_live=FakeLivePrice({"BTCUSDT": 90.0}))
r = c.get("/paper/crypto")
assert r.status_code == 200
assert "120,000" in r.text # compounded record headline
assert "marked to live prices now" in r.text # the live mark, clearly labeled secondary
def test_paper_handler_does_no_heavy_compute_only_light_nav_read() -> None:
# The /paper handler must source the headline + curve from a LIGHT persisted nav read (one indexed
# query), NEVER any heavy per-coin / warehouse compute (the 4Gi dashboard OOMs on that). Assert the
# bybit repo's nav_history is the read used, and no warehouse/per-coin path is hit, by spying on the repo.
bybit = PaperRepo("sqlite://")
bybit.migrate()
_seed_nav(bybit, n=120, start=100_000.0, end=200_000.0)
calls = {"nav_history": 0}
orig = bybit.nav_history
def _spy() -> list:
calls["nav_history"] += 1
return orig()
bybit.nav_history = _spy # type: ignore[method-assign]
c = _client_bybit_repo(bybit, bybit_live=FakeLivePrice({}))
r = c.get("/paper/crypto")
assert r.status_code == 200
assert calls["nav_history"] >= 1 # the light persisted read happened
assert _svg_point_count(r.text) > 2 # and drove the real curve
def test_paper_is_relative_url_leak_clean() -> None:
import re
abs_url = re.compile(r"https?://(?!www\.w3\.org/)[^\s\"'<>]+")
c = _client_with_forward(_seeded_bybit_paper_repo(), with_track=True)
r = c.get("/paper/crypto")
assert r.status_code == 200
hits = abs_url.findall(r.text)
assert not hits, f"leak-clean violation: {hits}"
# --- SSOT rollout: /paper is the Overview's visual language, single-venue ------------------------------
# The Paper page rolls the same SSOT pattern the Overview leads with: the SAME `headline_card` (connected
# backfill record + live-gate chip), then the unique live per-symbol book + trades + the individual
# constituent edges, all PAPER-badged. The walls of muted prose are gone.
def _engineered_headline_nav() -> list[float]:
"""A deterministic bybit backfill equity series whose nav_backfill_stats render the canonical headline
figures EXACTLY: final $394,085 / +294% total return / Sharpe 1.45 (annualised, the repo convention).
Alternating up/down daily returns (mean/std fixed for Sharpe), then rescaled so the final lands on the
headline equity. Pure arithmetic — same maths nav_backfill_stats runs."""
m, d, g = 309, 0.03955, 0.003001845
nav = [100_000.0]
for _ in range(m):
nav.append(nav[-1] * (1 + g + d))
nav.append(nav[-1] * (1 + g - d))
k = 394_085.0 / nav[-1]
return [x * k for x in nav]
def _seed_engineered_headline(repo: PaperRepo) -> None:
at = dt.datetime(2026, 6, 24, tzinfo=dt.UTC)
base = dt.date(2021, 1, 1)
for i, eq in enumerate(_engineered_headline_nav()):
repo.upsert_nav((base + dt.timedelta(days=i)).isoformat(), capital=100_000.0,
realized=0.0, unrealized=0.0, equity=eq, at=at)
def test_paper_bybit_renders_headline_card_and_keeps_positions_trades_prose_gone() -> None:
# The page leads with the SAME headline_card the Overview uses (exact backfill record, Sharpe to
# 2 decimals with NO tilde), carries a PAPER exec badge, KEEPS the unique live positions + trades, and
# the old walls of muted prose are GONE.
paper_repo_for_edges = _seeded_bybit_paper_repo() # carries bybit_sleeve_ret for the constituent edges
bybit = PaperRepo("sqlite://")
bybit.migrate()
# A live book (positions + trades) at the backfill's FIRST date, then overwrite that date's nav with the
# engineered backfill series so nav_history == the engineered series exactly (the headline maths is clean).
PaperBookService(bybit, capital=100_000.0).derive_and_persist(
run_date="2021-01-01", sleeve_weights={"positioning": 1.0},
symbol_weights_by_sleeve={"positioning": {"BTCUSDT": -0.01}},
prices={"BTCUSDT": 50_000.0}, at=dt.datetime(2026, 6, 24, tzinfo=dt.UTC))
_seed_engineered_headline(bybit)
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
fwd = ForwardNavRepo("sqlite://")
fwd.migrate()
c = TestClient(create_app(fwd, paper_repo=paper_repo_for_edges,
bybit_paper_repo=bybit, bybit_live=FakeLivePrice({"BTCUSDT": 50_000.0})))
r = c.get("/paper/crypto")
assert r.status_code == 200
# The headline_card (the SAME component the Overview leads with), exact figures.
assert "headline" in r.text # the headline_card wrapper class
assert "$394,085" in r.text
assert "+294%" in r.text
assert "Sharpe 1.45" in r.text # 2 decimals, the macro format
assert "~1.45" not in r.text and "~$394,085" not in r.text # NO tilde — it is the real record
# PAPER exec badge on the deploy book.
assert "badge-exec paper" in r.text
# The unique /paper detail is KEPT: the live per-symbol book + recent trades.
assert "open positions" in r.text
assert "recent trades" in r.text
assert "BTCUSDT" in r.text
# The old walls of muted prose are GONE.
assert "derived from the daily bybit book rebalance" not in r.text
assert "precomputed nightly from the survivorship-free Bybit feature store" not in r.text
assert "backtest only" not in r.text
def test_paper_is_the_deploy_section_only() -> None:
# Single-venue page: only the deploy section wrapper renders (the dimmed research wrapper the old
# Binance branch used is gone from this page).
c = _client_with_forward(_seeded_bybit_paper_repo(), with_track=True)
r = c.get("/paper/crypto")
assert r.status_code == 200
assert "section deploy" in r.text
assert "section research" not in r.text
assert "badge-exec paper" in r.text
Reference in New Issue View Git Blame Copy Permalink