124 lines
5.7 KiB
Python
124 lines
5.7 KiB
Python
"""DashboardService turns DB rows into view-models: a fleet overview (registry joined to summary) and a
|
|
per-strategy detail (summary + nav history)."""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
|
|
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
|
|
from fxhnt.application.dashboard_service import DashboardService
|
|
from fxhnt.application.forward_models import BacktestSummary
|
|
from fxhnt.application.forward_models import ForwardNavRow as Row
|
|
from fxhnt.application.forward_models import ForwardSummary
|
|
from fxhnt.registry import STRATEGY_REGISTRY
|
|
|
|
|
|
def _seeded() -> ForwardNavRepo:
|
|
repo = ForwardNavRepo("sqlite://")
|
|
repo.migrate()
|
|
at = dt.datetime(2026, 6, 14, 23, 30)
|
|
repo.upsert_rows([Row("unlock", "2026-06-05", 0.01, 1.01),
|
|
Row("unlock", "2026-06-06", 0.0099, 1.0201)], at)
|
|
repo.upsert_summary(ForwardSummary("unlock", "2026-06-06", 2, 1.0201, 0.0201, 1.2, -0.01),
|
|
"WAIT", "2/20 forward days", at)
|
|
# a surviving track (xsfunding) carries a backtest verdict; a RETIRED one (eqfactor_long) has a
|
|
# leftover backtest-summary DB row but must NOT surface (fleet is registry-driven).
|
|
repo.upsert_backtest_summary(BacktestSummary(
|
|
"xsfunding", "2026-05-30", cagr=0.12, ann_vol=0.18, sharpe=0.85, max_drawdown=-0.22,
|
|
passed=True, dsr=0.61, is_sharpe=0.9, oos_sharpe=0.7, pvalue=0.03), at)
|
|
repo.upsert_backtest_summary(BacktestSummary(
|
|
"eqfactor_long", "2026-05-30", cagr=-0.01, ann_vol=0.11, sharpe=-0.05, max_drawdown=-0.31,
|
|
passed=False, dsr=-0.2, is_sharpe=0.1, oos_sharpe=-0.3, pvalue=0.6), at)
|
|
return repo
|
|
|
|
|
|
def test_fleet_joins_registry_to_summary() -> None:
|
|
svc = DashboardService(_seeded())
|
|
fleet = svc.fleet()
|
|
unlock = next(f for f in fleet if f.strategy_id == "unlock")
|
|
assert unlock.display_name.startswith("Token-unlock dilution shorts")
|
|
assert unlock.days == 2 and unlock.gate_status == "WAIT"
|
|
assert round(unlock.total_return_pct, 2) == 2.01
|
|
|
|
|
|
def test_fleet_includes_registry_entries_without_data() -> None:
|
|
svc = DashboardService(_seeded())
|
|
fleet = svc.fleet()
|
|
funding = next(f for f in fleet if f.strategy_id == "xsfunding")
|
|
assert funding.days == 0 and funding.gate_status == "WAIT" # no data yet -> shown, not errored
|
|
|
|
|
|
def test_detail_returns_summary_and_history() -> None:
|
|
svc = DashboardService(_seeded())
|
|
detail = svc.detail("unlock")
|
|
assert detail is not None
|
|
assert detail.display_name.startswith("Token-unlock dilution shorts")
|
|
assert [p.date for p in detail.history] == ["2026-06-05", "2026-06-06"]
|
|
|
|
|
|
def test_detail_includes_raw_period_aggregations() -> None:
|
|
svc = DashboardService(_seeded())
|
|
detail = svc.detail("unlock")
|
|
assert set(detail.periods) == {"daily", "weekly", "monthly"}
|
|
assert [p.label for p in detail.periods["daily"]] == ["2026-06-05", "2026-06-06"]
|
|
assert abs(detail.periods["daily"][0].ret - 0.01) < 1e-9 # raw, not normalized
|
|
assert all(p.label.startswith("2026") for p in detail.periods["monthly"])
|
|
|
|
|
|
def test_detail_unknown_strategy_is_none() -> None:
|
|
svc = DashboardService(_seeded())
|
|
assert svc.detail("does-not-exist") is None
|
|
|
|
|
|
def test_fleet_surfaces_backtest_verdict() -> None:
|
|
fleet = DashboardService(_seeded()).fleet()
|
|
el = next(f for f in fleet if f.strategy_id == "xsfunding")
|
|
assert el.bt_status == "PASS"
|
|
assert abs(el.bt_sharpe - 0.85) < 1e-9
|
|
assert abs(el.bt_dsr - 0.61) < 1e-9
|
|
assert abs(el.bt_cagr - 0.12) < 1e-9
|
|
assert abs(el.bt_maxdd - (-0.22)) < 1e-9
|
|
assert el.bt_as_of == "2026-05-30"
|
|
|
|
# eqfactor_long is retired from the registry (unproven); even though a backtest-summary row still
|
|
# exists in the DB, the fleet is registry-driven so the retired track is no longer surfaced.
|
|
assert all(f.strategy_id != "eqfactor_long" for f in fleet)
|
|
|
|
|
|
def test_fleet_no_backtest_renders_blank() -> None:
|
|
fleet = DashboardService(_seeded()).fleet()
|
|
unlock = next(f for f in fleet if f.strategy_id == "unlock")
|
|
assert unlock.bt_status == ""
|
|
assert unlock.bt_sharpe is None and unlock.bt_dsr is None
|
|
assert unlock.bt_cagr is None and unlock.bt_maxdd is None
|
|
assert unlock.bt_as_of == ""
|
|
|
|
|
|
def test_archived_sleeves_hidden_everywhere(monkeypatch) -> None:
|
|
# A SYNTHETIC archived registry entry (`STRATEGY_REGISTRY` `archived: True` — the pattern
|
|
# `test_paper_books_web.py` uses for its synthetic `archivedstrat`/`newstrat`, in place of the deleted
|
|
# vrp-based test) must be excluded from every cockpit surface `ForwardNavRepo.registry()` feeds:
|
|
# `DashboardService.fleet()` and `.detail()`. Monkeypatch BEFORE `migrate()` so `_seed_registry()` (which
|
|
# seeds the DB registry table FROM `STRATEGY_REGISTRY` and prunes anything not present there) picks up
|
|
# the synthetic entry.
|
|
monkeypatch.setitem(STRATEGY_REGISTRY, "archivedstrat",
|
|
{"archived": True, "display_name": "Options income (put spreads)",
|
|
"sleeve": "equity-vol", "gate_spec": {}})
|
|
repo = _seeded() # seeds registry AFTER the monkeypatch above, so archivedstrat lands in the DB row set
|
|
svc = DashboardService(repo)
|
|
|
|
# registry() itself: the archived row never comes back, a non-archived row (unlock) does.
|
|
reg_ids = {r.strategy_id for r in repo.registry()}
|
|
assert "archivedstrat" not in reg_ids
|
|
assert "unlock" in reg_ids
|
|
|
|
# fleet(): archived sleeve excluded; a live sleeve still shows.
|
|
fleet = svc.fleet()
|
|
assert all(f.strategy_id != "archivedstrat" for f in fleet)
|
|
assert any(f.strategy_id == "unlock" for f in fleet)
|
|
|
|
# detail(): direct hit on the archived id returns None (404 at the route layer); a live id still resolves.
|
|
assert svc.detail("archivedstrat") is None
|
|
assert svc.detail("unlock") is not None
|
|
|
|
|