gitadmin/fxhnt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge feat/portfolio-backtester: portfolio backtester cockpit surface

Browse Source 
/paper/portfolio — assemble edges + weighting method + date range → portfolio
equity curve, full/left-tail correlation heatmap, marginal-Sharpe table. Reuses
the existing combine_book/combine_returns/compute_stats/marginal/correlation engine;
Plotly vendored to /static (leak-clean, no CDN). Slice 1 of cockpit-as-platform.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Jeroen Grusewski
2026-07-21 19:16:34 +00:00
parent 0d8ce554a3 c51e45806c
commit d58c8dc04d
10 changed files with 1360 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

750
docs/superpowers/plans/2026-07-21-portfolio-backtester.md Normal file
View File

@@ -0,0 +1,750 @@
# Portfolio Backtester Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** A new `/paper/portfolio` cockpit surface where the operator selects an arbitrary set of edges, a weighting method, and a date range, and gets a portfolio equity curve + Sharpe/CAGR/maxDD **plus** the full/left-tail correlation matrix and each edge's marginal-Sharpe contribution — reusing the existing combiner/metrics/diversification engine end-to-end.
**Architecture:** Three thin layers on top of a production-tested engine. (1) A cross-store **series aggregator** unions `bybit_sleeve_returns()` + `sim_curve_returns()` into the one `series_by_edge: dict[str, dict[int,float]]` shape the engine already consumes. (2) A pure **report bundler** slices to a date range, runs `combine_book` (default) or `combine_returns` (explicit-weights what-if), then `compute_stats`, `full_correlation`/`left_tail_correlation`, and `marginal_sharpe` on the aligned arrays, returning one report dict. (3) A FastAPI **route + template** (`/paper/portfolio` page + `/paper/portfolio/run` HTMX partial) rendering the curve + heatmap via Plotly (CDN) and the marginal-Sharpe table as HTML. Pure additive + read-only.
**Tech Stack:** Python 3.12, FastAPI + Jinja2 + HTMX, NumPy, existing `fxhnt` domain/application engine, Plotly (browser-side, CDN).
## Global Constraints
- **Read-only + additive.** No task may touch any live-fund path, the execution CLI, `/paper/sim`, or any existing route/template/service method. Only NEW files + NEW route handlers + NEW nav links + one NEW `<script>` in `base.html`.
- **No recompute on the request path.** The route reads ONLY the precomputed stored series (`bybit_sleeve_returns`, `sim_curve_returns`) — never triggers an edge/replay computation. Mirrors the `/paper/sim` "reads the honest precomputed curve" rule.
- **Engine is reused verbatim — do NOT rebuild.** `align_edges`/`combine_book` (`src/fxhnt/application/book_allocator.py`), `combine_returns` (`src/fxhnt/domain/portfolio/combine.py`), `compute_stats` (`src/fxhnt/domain/backtest.py`), `marginal_sharpe` (`src/fxhnt/domain/diversification/marginal.py`), `full_correlation`/`left_tail_correlation` (`src/fxhnt/domain/diversification/correlation.py`). No copies, no re-derivations.
- **Exact epoch-day axis.** All stored series are keyed by `epoch_day = date.toordinal()` (NOT Unix days). `combine_book` returns `{epoch_day: ret}`. Convert to ISO with `datetime.date.fromordinal(epoch_day).isoformat()`. `combine_returns` uses a DIFFERENT axis (date-STRING tuples) — the bundler bridges the two explicitly (Task 2).
- **Graceful, never 500.** A missing/empty series, an empty selection, or an unparseable weight must degrade to an empty/pending report, never a 500 — matching the `# a missing/empty nav table must never 500` guard style in `app.py`.
- **Available edges (v1 fixed set):** `crypto_tstrend`, `unlock`, `xsfunding`, `positioning` (from `bybit_sleeve_returns()`), and `multistrat` (from `sim_curve_returns("multistrat")`). `multistrat_levered` is available from `sim_curve_returns` but is observe-only — include it as a selectable but not default-on.
- **Plotly loads from CDN in the browser** (`https://cdn.plot.ly/plotly-2.35.2.min.js`). This is a BROWSER fetch over the tailscale-served page, NOT pod egress — the vendored-htmx "no CDN / tailnet egress" comment in `base.html` is about the POD's network policy and does NOT apply to the operator's browser. Add a one-line comment saying so next to the new `<script>`.
- **Tests:** `pytest -n auto` is the default (`pyproject.toml`). Run individual tests with `-p no:xdist` implicitly disabled by naming the node — use `pytest tests/path::test_name -p no:cacheprovider` is NOT needed; plain `pytest tests/path::test_name` works. Web tests build the app in-memory via `create_app(...)` with SQLite `PaperRepo`/`ForwardNavRepo` + `fastapi.testclient.TestClient`.
---
## File Structure
- **Create** `src/fxhnt/application/portfolio_backtest.py` — the aggregator + report bundler (pure application logic, no web, no I/O beyond the two repo accessors passed in). Two public functions: `aggregate_edge_series(repo, edges)` and `build_portfolio_report(repo, *, edges, method, weights, start, end)`.
- **Create** `tests/application/test_portfolio_backtest.py` — unit tests for both functions.
- **Modify** `src/fxhnt/adapters/web/app.py` — add `GET /paper/portfolio` (form page) + `GET /paper/portfolio/run` (HTMX partial) handlers, and the module-level list of selectable edges. NEW handlers only; no existing handler touched.
- **Create** `src/fxhnt/adapters/web/templates/portfolio.html` — the page: edge multiselect + method toggle + date range + `hx-get` trigger.
- **Create** `src/fxhnt/adapters/web/templates/_portfolio_result.html` — the HTMX partial: Plotly equity curve `<div>` + Plotly heatmap `<div>` + marginal-Sharpe HTML table + headline metrics.
- **Modify** `src/fxhnt/adapters/web/templates/base.html` — add the Plotly `<script>` (CDN) in `<head>`, and a "Portfolio" nav link to BOTH the desktop `.bar` nav and the mobile `.tabs` nav.
- **Create** `tests/integration/test_portfolio_web.py` — web smoke: page renders, run partial returns curve+metrics+heatmap+marginal table, empty selection degrades gracefully, explicit-weights mode works.
---
## Task 1: Cross-store series aggregator
**Files:**
- Create: `src/fxhnt/application/portfolio_backtest.py`
- Test: `tests/application/test_portfolio_backtest.py`
**Interfaces:**
- Consumes: `PaperRepo.bybit_sleeve_returns() -> dict[str, dict[int,float]]` and `PaperRepo.sim_curve_returns(strategy_id: str) -> dict[int,float]` (both in `src/fxhnt/adapters/persistence/paper_repo.py`, already exist).
- Produces: `aggregate_edge_series(repo, edges: list[str]) -> dict[str, dict[int,float]]` — the `series_by_edge` shape `align_edges`/`combine_book` consume. Also produces the module constants `BYBIT_EDGES` and `SIM_EDGES` used by later tasks.
- [ ] **Step 1: Write the failing test**
```python
# tests/application/test_portfolio_backtest.py
"""Unit tests for the portfolio-backtest aggregator + report bundler (read-only, pure)."""
from __future__ import annotations
import datetime as dt
from fxhnt.application.portfolio_backtest import aggregate_edge_series
class _FakeRepo:
"""Minimal stand-in exposing only the two accessors the aggregator reads."""
def __init__(self, bybit: dict[str, dict[int, float]], sim: dict[str, dict[int, float]]):
self._bybit = bybit
self._sim = sim
def bybit_sleeve_returns(self) -> dict[str, dict[int, float]]:
return self._bybit
def sim_curve_returns(self, strategy_id: str) -> dict[int, float]:
return self._sim.get(strategy_id, {})
def _d(y: int, m: int, day: int) -> int:
return dt.date(y, m, day).toordinal()
def test_aggregate_unions_bybit_and_sim_sources():
repo = _FakeRepo(
bybit={
"crypto_tstrend": {_d(2026, 1, 1): 0.01, _d(2026, 1, 2): -0.02},
"unlock": {_d(2026, 1, 1): 0.003},
},
sim={"multistrat": {_d(2026, 1, 1): 0.005, _d(2026, 1, 2): 0.004}},
)
out = aggregate_edge_series(repo, ["crypto_tstrend", "multistrat"])
assert set(out) == {"crypto_tstrend", "multistrat"}
assert out["crypto_tstrend"][_d(2026, 1, 2)] == -0.02
assert out["multistrat"][_d(2026, 1, 1)] == 0.005
def test_aggregate_skips_unknown_and_empty_edges():
repo = _FakeRepo(bybit={"unlock": {_d(2026, 1, 1): 0.003}}, sim={})
out = aggregate_edge_series(repo, ["unlock", "does_not_exist", "multistrat"])
assert set(out) == {"unlock"} # unknown + empty-series edges dropped
```
- [ ] **Step 2: Run test to verify it fails**
Run: `pytest tests/application/test_portfolio_backtest.py -v`
Expected: FAIL — `ModuleNotFoundError: No module named 'fxhnt.application.portfolio_backtest'`
- [ ] **Step 3: Write minimal implementation**
```python
# src/fxhnt/application/portfolio_backtest.py
"""Portfolio backtester (cockpit slice 1): assemble an arbitrary set of edges from the PRECOMPUTED stored
return series, run them through the EXISTING book combiner / metrics / diversification engine, and return one
report the /paper/portfolio surface renders. Pure application logic — reads only the two repo accessors, never
recomputes an edge/replay on the request path. Read-only; cannot touch the live fund."""
from __future__ import annotations
from typing import Protocol
# The selectable edges, grouped by their stored-series source (v1 fixed set).
BYBIT_EDGES: tuple[str, ...] = ("crypto_tstrend", "unlock", "xsfunding", "positioning")
SIM_EDGES: tuple[str, ...] = ("multistrat", "multistrat_levered")
ALL_EDGES: tuple[str, ...] = BYBIT_EDGES + SIM_EDGES
class _SeriesRepo(Protocol):
def bybit_sleeve_returns(self) -> dict[str, dict[int, float]]: ...
def sim_curve_returns(self, strategy_id: str) -> dict[int, float]: ...
def aggregate_edge_series(repo: _SeriesRepo, edges: list[str]) -> dict[str, dict[int, float]]:
"""Union the selected edges' PRECOMPUTED per-day series into one `{edge: {epoch_day: ret}}` map — the
exact shape `book_allocator.align_edges`/`combine_book` consume. Bybit sleeves come from the one
`bybit_sleeve_returns()` read; each IBKR recompute-replay book from its own `sim_curve_returns(id)` read.
Unknown edges and edges whose stored series is empty are silently dropped (a selection referencing a
not-yet-precomputed book must degrade, never error)."""
bybit = repo.bybit_sleeve_returns()
out: dict[str, dict[int, float]] = {}
for e in edges:
if e in BYBIT_EDGES:
series = bybit.get(e) or {}
elif e in SIM_EDGES:
series = repo.sim_curve_returns(e)
else:
continue # unknown edge name — drop
if series:
out[e] = dict(series)
return out
```
- [ ] **Step 4: Run test to verify it passes**
Run: `pytest tests/application/test_portfolio_backtest.py -v`
Expected: PASS (2 passed)
- [ ] **Step 5: Commit**
```bash
git add src/fxhnt/application/portfolio_backtest.py tests/application/test_portfolio_backtest.py
git commit -m "feat(portfolio): cross-store edge-series aggregator"
```
---
## Task 2: Report bundler
**Files:**
- Modify: `src/fxhnt/application/portfolio_backtest.py`
- Test: `tests/application/test_portfolio_backtest.py`
**Interfaces:**
- Consumes: `aggregate_edge_series` (Task 1); `align_edges`/`combine_book` from `fxhnt.application.book_allocator`; `combine_returns` from `fxhnt.domain.portfolio.combine`; `compute_stats` from `fxhnt.domain.backtest`; `marginal_sharpe` from `fxhnt.domain.diversification.marginal`; `full_correlation`/`left_tail_correlation` from `fxhnt.domain.diversification.correlation`.
- Engine facts (verbatim, do not re-derive): `combine_book(series_by_edge, *, vol_lookback=60, target_vol=0.12, kelly_fraction=0.5, max_sleeve_weight=0.4, periods_per_year=365, rebalance_every=21, min_sleeve_sharpe=None, marginal_gate=False) -> dict[int,float]`. `align_edges(series_by_edge) -> (dates: list[int], arrs: dict[str, np.ndarray])`. `combine_returns(sleeves: list[tuple[float, tuple[str,...], np.ndarray]]) -> (dates: tuple[str,...], np.ndarray)`. `compute_stats(returns: np.ndarray, periods_per_year: int = 252) -> BacktestStats` (dataclass; access its fields — inspect the dataclass in `src/fxhnt/domain/backtest.py` and surface CAGR, Sharpe, max drawdown, vol). `marginal_sharpe(book: np.ndarray, sleeve: np.ndarray, *, weight=0.2) -> float`. `full_correlation(a, b) -> float`; `left_tail_correlation(sleeve, book, *, tail_frac=0.10) -> float`.
- Produces: `build_portfolio_report(repo, *, edges: list[str], method: str = "book", weights: dict[str,float] | None = None, start: str | None = None, end: str | None = None) -> dict` — the report dict later tasks render. Keys: `edges` (list[str], the ones actually used), `method`, `dates_iso` (list[str]), `equity` (list[float], compounded from $1.0), `returns` (list[float]), `metrics` (dict: `cagr`,`sharpe`,`max_drawdown`,`vol`,`n_days`), `correlation` (list of `{a,b,full,tail}` for each unordered pair), `marginal` (list of `{edge, marginal_sharpe}`), `empty` (bool — True when no usable edge/overlap).
- [ ] **Step 1: Write the failing test**
```python
# append to tests/application/test_portfolio_backtest.py
import math
from fxhnt.application.portfolio_backtest import build_portfolio_report
def _series(start_ord: int, rets: list[float]) -> dict[int, float]:
return {start_ord + i: r for i, r in enumerate(rets)}
def _repo_with_overlap() -> "_FakeRepo":
s = _d(2026, 1, 1)
# 40 days so trailing windows in combine_book/left-tail have room.
a = [0.01 if i % 2 else -0.008 for i in range(40)]
b = [0.006 if i % 3 else -0.004 for i in range(40)]
c = [0.004 for _ in range(40)]
return _FakeRepo(
bybit={"crypto_tstrend": _series(s, a), "unlock": _series(s, b)},
sim={"multistrat": _series(s, c)},
)
def test_report_book_method_has_curve_metrics_corr_marginal():
rep = build_portfolio_report(
_repo_with_overlap(), edges=["crypto_tstrend", "unlock", "multistrat"], method="book"
)
assert rep["empty"] is False
assert rep["method"] == "book"
assert set(rep["edges"]) == {"crypto_tstrend", "unlock", "multistrat"}
assert len(rep["dates_iso"]) == len(rep["equity"]) == len(rep["returns"])
assert rep["equity"][0] > 0.0 # compounded from a positive base
for k in ("cagr", "sharpe", "max_drawdown", "vol", "n_days"):
assert k in rep["metrics"]
# 3 edges -> 3 unordered pairs.
assert len(rep["correlation"]) == 3
for pair in rep["correlation"]:
assert set(pair) == {"a", "b", "full", "tail"}
assert len(rep["marginal"]) == 3
assert all(set(m) == {"edge", "marginal_sharpe"} for m in rep["marginal"])
def test_report_explicit_weights_method_uses_supplied_weights():
rep = build_portfolio_report(
_repo_with_overlap(),
edges=["crypto_tstrend", "unlock"],
method="explicit",
weights={"crypto_tstrend": 0.7, "unlock": 0.3},
)
assert rep["empty"] is False
assert rep["method"] == "explicit"
assert len(rep["returns"]) > 0
def test_report_empty_when_no_edges_selected():
rep = build_portfolio_report(_repo_with_overlap(), edges=[], method="book")
assert rep["empty"] is True
assert rep["equity"] == []
assert rep["correlation"] == []
def test_report_date_range_windows_the_series():
repo = _repo_with_overlap()
full = build_portfolio_report(repo, edges=["crypto_tstrend", "unlock"], method="book")
windowed = build_portfolio_report(
repo, edges=["crypto_tstrend", "unlock"], method="book",
start="2026-01-10", end="2026-01-20",
)
assert len(windowed["dates_iso"]) < len(full["dates_iso"])
assert windowed["dates_iso"][0] >= "2026-01-10"
assert windowed["dates_iso"][-1] <= "2026-01-20"
```
- [ ] **Step 2: Run test to verify it fails**
Run: `pytest tests/application/test_portfolio_backtest.py -v`
Expected: FAIL — `ImportError: cannot import name 'build_portfolio_report'`
- [ ] **Step 3: Write minimal implementation**
Append to `src/fxhnt/application/portfolio_backtest.py`. First inspect `compute_stats` / `BacktestStats` field names in `src/fxhnt/domain/backtest.py` and map them to `cagr`/`sharpe`/`max_drawdown`/`vol` (use the actual dataclass attribute names — do not invent). The code below assumes `BacktestStats` exposes `cagr`, `sharpe`, `max_drawdown`, `vol`; adjust the four attribute reads in `_metrics_from_stats` if the real names differ.
```python
import datetime as dt
from itertools import combinations
import numpy as np
from fxhnt.application.book_allocator import align_edges, combine_book
from fxhnt.domain.backtest import compute_stats
from fxhnt.domain.diversification.correlation import full_correlation, left_tail_correlation
from fxhnt.domain.diversification.marginal import marginal_sharpe
from fxhnt.domain.portfolio.combine import combine_returns
_PERIODS_PER_YEAR = 365 # stored series are calendar-daily (matches combine_book's default)
def _window(series_by_edge: dict[str, dict[int, float]], start: str | None, end: str | None
) -> dict[str, dict[int, float]]:
"""Clip every edge's {epoch_day: ret} to the inclusive [start, end] ISO date range (either bound optional)."""
lo = dt.date.fromisoformat(start).toordinal() if start else None
hi = dt.date.fromisoformat(end).toordinal() if end else None
out: dict[str, dict[int, float]] = {}
for e, s in series_by_edge.items():
clipped = {d: r for d, r in s.items()
if (lo is None or d >= lo) and (hi is None or d <= hi)}
if clipped:
out[e] = clipped
return out
def _empty_report(method: str) -> dict:
return {"edges": [], "method": method, "dates_iso": [], "equity": [], "returns": [],
"metrics": {"cagr": 0.0, "sharpe": 0.0, "max_drawdown": 0.0, "vol": 0.0, "n_days": 0},
"correlation": [], "marginal": [], "empty": True}
def _metrics_from_stats(returns: np.ndarray) -> dict:
if returns.size == 0:
return {"cagr": 0.0, "sharpe": 0.0, "max_drawdown": 0.0, "vol": 0.0, "n_days": 0}
st = compute_stats(returns, periods_per_year=_PERIODS_PER_YEAR)
return {"cagr": float(st.cagr), "sharpe": float(st.sharpe),
"max_drawdown": float(st.max_drawdown), "vol": float(st.vol), "n_days": int(returns.size)}
def _equity_curve(returns: np.ndarray) -> list[float]:
"""Compound daily returns from a $1.0 base (the surface scales by capital client-side / in the template)."""
eq, curve = 1.0, []
for r in returns:
eq *= (1.0 + float(r))
curve.append(eq)
return curve
def build_portfolio_report(repo: _SeriesRepo, *, edges: list[str], method: str = "book",
weights: dict[str, float] | None = None,
start: str | None = None, end: str | None = None) -> dict:
"""Assemble the full portfolio report for the selected edges + weighting method + date range.
method="book" -> the live inverse-vol/cap/vol-target+Kelly allocator (`combine_book`) — "what does my
real allocation do with these edges?".
method="explicit" -> a plain capital-weighted blend of the supplied per-edge `weights` (`combine_returns`) —
"what if positioning were 40%?".
The equity/metrics reflect the chosen weighting; the correlation matrix (full AND left-tail) and per-edge
marginal-Sharpe are computed on the RAW aligned edge arrays (properties of the edges, weight-independent).
Read-only; never recomputes an edge."""
series = _window(aggregate_edge_series(repo, edges), start, end)
if not series:
return _empty_report(method)
# RAW aligned per-edge arrays on the shared epoch-day axis (0.0-fill on flat days) — the basis for the
# book curve's diagnostics (correlation + marginal-Sharpe).
epoch_days, arrs = align_edges(series)
if not epoch_days:
return _empty_report(method)
used = list(arrs)
if method == "explicit" and weights:
# combine_returns aligns on the INTERSECTION of date-STRING axes; build one Sleeve per used edge.
iso = [dt.date.fromordinal(d).isoformat() for d in epoch_days]
sleeves = [(float(weights.get(e, 0.0)), tuple(iso), arrs[e]) for e in used]
_cdates, book_ret = combine_returns(sleeves)
# combine_returns may drop to the intersection; re-derive the ISO axis from its returned dates.
dates_iso = list(_cdates)
returns = np.asarray(book_ret, dtype=float)
else:
method = "book"
book_map = combine_book(series) # {epoch_day: book_return}
book_days = sorted(book_map)
dates_iso = [dt.date.fromordinal(d).isoformat() for d in book_days]
returns = np.array([book_map[d] for d in book_days], dtype=float)
if returns.size == 0:
return _empty_report(method)
# Diagnostics on the raw edge arrays vs the book array (aligned to min length inside the domain fns).
book_arr = returns
correlation = [{"a": a, "b": b,
"full": _nan_to_none(full_correlation(arrs[a], arrs[b])),
"tail": _nan_to_none(left_tail_correlation(arrs[a], book_arr) if a == b else
left_tail_correlation(arrs[b], book_arr))}
for a, b in combinations(used, 2)]
marginal = [{"edge": e, "marginal_sharpe": _nan_to_none(marginal_sharpe(book_arr, arrs[e]))}
for e in used]
return {"edges": used, "method": method, "dates_iso": dates_iso,
"equity": _equity_curve(returns), "returns": [float(r) for r in returns],
"metrics": _metrics_from_stats(returns), "correlation": correlation,
"marginal": marginal, "empty": False}
def _nan_to_none(x: float) -> float | None:
"""JSON/template-safe: the domain fns return NaN when a series is too short to estimate; surface as None."""
return None if x is None or (isinstance(x, float) and np.isnan(x)) else float(x)
```
NOTE on the `correlation` pair `tail` field: `left_tail_correlation(sleeve, book)` measures a sleeve vs the book on the book's worst days — it is naturally per-edge, not per-pair. For the pairwise heatmap, the honest per-PAIR tail measure is the left-tail correlation of the two edges conditioned on one being in its worst tail. Use the pair's tail co-movement: replace the `tail` computation with `left_tail_correlation(arrs[a], arrs[b])` (edge-a's tail vs edge-b) so each cell is a true pairwise number. Fix the `_window`/report code accordingly:
```python
correlation = [{"a": a, "b": b,
"full": _nan_to_none(full_correlation(arrs[a], arrs[b])),
"tail": _nan_to_none(left_tail_correlation(arrs[a], arrs[b]))}
for a, b in combinations(used, 2)]
```
(Use this pairwise form — drop the `a == b` branch above.)
- [ ] **Step 4: Run test to verify it passes**
Run: `pytest tests/application/test_portfolio_backtest.py -v`
Expected: PASS (6 passed). If `BacktestStats` field names differ, the failure will be an `AttributeError` in `_metrics_from_stats` — read the dataclass and fix the four reads, then re-run.
- [ ] **Step 5: Commit**
```bash
git add src/fxhnt/application/portfolio_backtest.py tests/application/test_portfolio_backtest.py
git commit -m "feat(portfolio): report bundler (book + explicit weights, corr + marginal-Sharpe)"
```
---
## Task 3: Route + template + Plotly rendering
**Files:**
- Modify: `src/fxhnt/adapters/web/app.py` (add two handlers + one module-level edge list; NO existing handler touched)
- Create: `src/fxhnt/adapters/web/templates/portfolio.html`
- Create: `src/fxhnt/adapters/web/templates/_portfolio_result.html`
- Modify: `src/fxhnt/adapters/web/templates/base.html` (Plotly `<script>` + "Portfolio" nav link in both navs)
- Test: `tests/integration/test_portfolio_web.py`
**Interfaces:**
- Consumes: `build_portfolio_report` + `ALL_EDGES`/`BYBIT_EDGES`/`SIM_EDGES` (Tasks 1–2); `create_app(...)` construction + `PaperRepo`/`ForwardNavRepo` + `TestClient` (pattern from `tests/integration/test_bybit_sim_web.py`); the existing `_paper_repo()` helper inside `app.py` (returns the injected/settings `PaperRepo`); `templates.TemplateResponse` (already bound in `app.py`).
- Produces: `GET /paper/portfolio` (page) + `GET /paper/portfolio/run?edges=...&method=...&start=...&end=...&w_<edge>=...` (HTMX partial). Query contract: `edges` repeated (`?edges=crypto_tstrend&edges=unlock`), `method` in `{book, explicit}`, `w_<edge>` float for explicit mode.
- [ ] **Step 1: Write the failing test**
```python
# tests/integration/test_portfolio_web.py
"""Web smoke for the portfolio backtester (/paper/portfolio): assemble edges from the PRECOMPUTED stored
series and render a portfolio curve + metrics + full/left-tail correlation heatmap + marginal-Sharpe table,
WITHOUT recomputing any edge on the request path. Read-only + additive: no live-fund path touched.
NO network — in-memory SQLite repos, seeded directly."""
from __future__ import annotations
import datetime as dt
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
def _seed(repo: PaperRepo) -> None:
at = dt.datetime(2026, 1, 1, tzinfo=dt.timezone.utc)
start = dt.date(2026, 1, 1)
for i in range(40):
d = (start + dt.timedelta(days=i)).isoformat()
repo.upsert_bybit_sleeve_ret("crypto_tstrend", d, 0.01 if i % 2 else -0.008, at=at)
repo.upsert_bybit_sleeve_ret("unlock", d, 0.006 if i % 3 else -0.004, at=at)
repo.upsert_sim_curve_ret("multistrat", d, 0.004, at=at)
def _client() -> TestClient:
paper = PaperRepo("sqlite://")
paper.migrate()
_seed(paper)
fwd = ForwardNavRepo("sqlite://")
fwd.migrate()
app = create_app(repo=fwd, bybit_paper_repo=paper, paper_repo=paper)
return TestClient(app)
def test_portfolio_page_renders_form():
r = _client().get("/paper/portfolio")
assert r.status_code == 200
body = r.text
assert "crypto_tstrend" in body and "multistrat" in body
assert "/paper/portfolio/run" in body # the HTMX target
def test_portfolio_run_book_method_returns_curve_metrics_corr_marginal():
r = _client().get("/paper/portfolio/run",
params=[("edges", "crypto_tstrend"), ("edges", "unlock"),
("edges", "multistrat"), ("method", "book")])
assert r.status_code == 200
body = r.text
assert "Plotly" in body # the curve/heatmap render scripts
assert "Sharpe" in body # headline metrics table
assert "Marginal" in body # marginal-Sharpe table heading
def test_portfolio_run_empty_selection_degrades_gracefully():
r = _client().get("/paper/portfolio/run", params={"method": "book"})
assert r.status_code == 200 # never a 500
assert "Select at least one edge" in r.text or "no edges" in r.text.lower()
def test_portfolio_run_explicit_weights_mode():
r = _client().get("/paper/portfolio/run",
params=[("edges", "crypto_tstrend"), ("edges", "unlock"),
("method", "explicit"), ("w_crypto_tstrend", "0.7"),
("w_unlock", "0.3")])
assert r.status_code == 200
assert "Plotly" in r.text
```
- [ ] **Step 2: Run test to verify it fails**
Run: `pytest tests/integration/test_portfolio_web.py -v`
Expected: FAIL — `/paper/portfolio` returns 404 (route not registered).
First confirm the `create_app` signature accepts `paper_repo=` — check with:
Run: `grep -n "def create_app" src/fxhnt/adapters/web/app.py`
If the kwarg differs (e.g. no separate `paper_repo`), adjust the test's `create_app(...)` call to the real signature (the Bybit repo is passed as `bybit_paper_repo=`; some apps read the sim series off the same repo). Use whatever `_paper_repo()` inside `app.py` resolves to as the report's repo.
- [ ] **Step 3: Add the module-level edge list + two route handlers**
In `src/fxhnt/adapters/web/app.py`, near the other `/paper/...` handlers (after `paper_sim_run`), add:
```python
@app.get("/paper/portfolio", response_class=HTMLResponse)
def paper_portfolio(request: Request) -> HTMLResponse:
"""Render the edge-multiselect + method toggle + date-range form INSTANTLY; the three-part report
(curve + full/left-tail correlation heatmap + marginal-Sharpe table) lazy-loads from
/paper/portfolio/run. Read-only: reads ONLY the precomputed stored series (no edge recompute)."""
from fxhnt.application.portfolio_backtest import BYBIT_EDGES, SIM_EDGES
# Pre-fill the date bounds from the union range of the available stored series so the pickers are
# never blank and can't go off-data.
repo = _paper_repo()
bybit = {}
try:
bybit = repo.bybit_sleeve_returns()
except Exception: # noqa: BLE001 — a missing table must never 500 the form
bybit = {}
all_days: set[int] = set()
for s in bybit.values():
all_days |= set(s)
try:
all_days |= set(repo.sim_curve_returns("multistrat"))
except Exception: # noqa: BLE001
pass
lo = dt.date.fromordinal(min(all_days)).isoformat() if all_days else ""
hi = dt.date.fromordinal(max(all_days)).isoformat() if all_days else ""
ctx = {"bybit_edges": BYBIT_EDGES, "sim_edges": SIM_EDGES,
"range_min": lo, "range_max": hi, "now": _now(),
"labels": {e: display_name(e) for e in (*BYBIT_EDGES, *SIM_EDGES)}}
return templates.TemplateResponse(request, "portfolio.html", ctx)
@app.get("/paper/portfolio/run", response_class=HTMLResponse)
def paper_portfolio_run(request: Request) -> HTMLResponse:
"""Build the report for the selected edges + method + range and render the three-part partial.
Empty selection / missing series degrade to a 'select an edge' / 'precomputing' note, never a 500."""
from fxhnt.application.portfolio_backtest import ALL_EDGES, build_portfolio_report
qp = request.query_params
edges = [e for e in qp.getlist("edges") if e in ALL_EDGES]
method = qp.get("method") if qp.get("method") in ("book", "explicit") else "book"
start = qp.get("start") or None
end = qp.get("end") or None
weights: dict[str, float] = {}
for e in edges:
raw = qp.get(f"w_{e}")
if raw:
try:
weights[e] = float(raw)
except ValueError:
pass
try:
report = build_portfolio_report(_paper_repo(), edges=edges, method=method,
weights=weights or None, start=start, end=end)
except Exception: # noqa: BLE001 — degrade to an empty report, never 500
report = {"empty": True, "edges": [], "method": method, "dates_iso": [], "equity": [],
"returns": [], "metrics": {}, "correlation": [], "marginal": []}
return templates.TemplateResponse(request, "_portfolio_result.html",
{"r": report, "selected": edges})
```
Verify `display_name`, `_now`, `_paper_repo`, `HTMLResponse`, `Request`, `dt` are all already imported/defined in `app.py` (they are used by neighboring handlers). If `display_name` isn't in scope at that point, import it the same way `_SIM_BOOK_LABELS` does.
- [ ] **Step 4: Create the page template**
```html
{# src/fxhnt/adapters/web/templates/portfolio.html #}
{% extends "base.html" %}
{% block title %}Portfolio · fxhnt{% endblock %}
{% block content %}
<section class="card">
<h2>Portfolio Backtest</h2>
<p class="muted">Assemble edges, pick a weighting method + date range, and see the portfolio-level
Sharpe/CAGR/maxDD plus the correlation (full &amp; left-tail) and marginal-Sharpe diagnostics —
all from the precomputed series (no recompute).</p>
<form hx-get="/paper/portfolio/run" hx-target="#portfolio-result" hx-trigger="submit"
hx-indicator="#portfolio-loading">
<fieldset>
<legend>Edges</legend>
{% for e in bybit_edges %}
<label class="chk"><input type="checkbox" name="edges" value="{{ e }}"> {{ labels[e] }}</label>
{% endfor %}
{% for e in sim_edges %}
<label class="chk"><input type="checkbox" name="edges" value="{{ e }}"> {{ labels[e] }}</label>
{% endfor %}
</fieldset>
<fieldset>
<legend>Method</legend>
<label class="chk"><input type="radio" name="method" value="book" checked> Book (live allocator)</label>
<label class="chk"><input type="radio" name="method" value="explicit"> Explicit weights</label>
</fieldset>
<fieldset id="explicit-weights">
<legend>Explicit weights (used only in “explicit” mode)</legend>
{% for e in bybit_edges %}
<label>{{ labels[e] }} <input type="number" step="0.05" min="0" max="1" name="w_{{ e }}"></label>
{% endfor %}
{% for e in sim_edges %}
<label>{{ labels[e] }} <input type="number" step="0.05" min="0" max="1" name="w_{{ e }}"></label>
{% endfor %}
</fieldset>
<fieldset>
<legend>Date range</legend>
<label>Start <input type="date" name="start" value="{{ range_min }}"
min="{{ range_min }}" max="{{ range_max }}"></label>
<label>End <input type="date" name="end" value="{{ range_max }}"
min="{{ range_min }}" max="{{ range_max }}"></label>
</fieldset>
<button type="submit">Run backtest</button>
<span id="portfolio-loading" class="htmx-indicator muted">running…</span>
</form>
<div id="portfolio-result"></div>
</section>
{% endblock %}
```
- [ ] **Step 5: Create the result partial (Plotly curve + heatmap + tables)**
```html
{# src/fxhnt/adapters/web/templates/_portfolio_result.html #}
{% if r.empty %}
<p class="muted">Select at least one edge and run the backtest.</p>
{% else %}
<div class="metrics-row">
<div class="metric"><span class="k">CAGR</span><span class="v">{{ '%.1f'|format(r.metrics.cagr * 100) }}%</span></div>
<div class="metric"><span class="k">Sharpe</span><span class="v">{{ '%.2f'|format(r.metrics.sharpe) }}</span></div>
<div class="metric"><span class="k">Max DD</span><span class="v">{{ '%.1f'|format(r.metrics.max_drawdown * 100) }}%</span></div>
<div class="metric"><span class="k">Vol</span><span class="v">{{ '%.1f'|format(r.metrics.vol * 100) }}%</span></div>
<div class="metric"><span class="k">Days</span><span class="v">{{ r.metrics.n_days }}</span></div>
</div>
<div id="pf-equity" class="chart"></div>
<h3>Correlation (full / left-tail)</h3>
<div id="pf-heatmap" class="chart"></div>
<h3>Marginal Sharpe (leave-one-out)</h3>
<table class="data">
<thead><tr><th>Edge</th><th>Marginal Sharpe</th></tr></thead>
<tbody>
{% for m in r.marginal %}
<tr><td data-label="Edge">{{ m.edge }}</td>
<td data-label="Marginal Sharpe">{{ '—' if m.marginal_sharpe is none else '%.3f'|format(m.marginal_sharpe) }}</td></tr>
{% endfor %}
</tbody>
</table>
{# Two server-rendered matrices (full + left-tail) → one heatmap with a Full/Left-tail toggle. Jinja renders
the matrices server-side (a JS closure can't parameterize a Jinja loop), so the JS is pure Plotly. #}
<script>
(function () {
var dates = {{ r.dates_iso|tojson }};
var equity = {{ r.equity|tojson }};
var edges = {{ r.marginal|map(attribute='edge')|list|tojson }};
Plotly.newPlot('pf-equity',
[{x: dates, y: equity, type: 'scatter', mode: 'lines', line: {color: '#4c9aff'}}],
{margin: {t: 10, r: 10, b: 30, l: 40}, paper_bgcolor: 'transparent',
plot_bgcolor: 'transparent', font: {color: '#c9d1d9'}, yaxis: {title: 'Equity (×)'}},
{displayModeBar: false, responsive: true});
var idx = {}; edges.forEach(function (e, i) { idx[e] = i; });
function blank() { return edges.map(function () { return edges.map(function () { return null; }); }); }
var full = blank(), tail = blank();
{% for p in r.correlation %}
full[idx[{{ p.a|tojson }}]][idx[{{ p.b|tojson }}]] = full[idx[{{ p.b|tojson }}]][idx[{{ p.a|tojson }}]] = {{ 'null' if p.full is none else p.full }};
tail[idx[{{ p.a|tojson }}]][idx[{{ p.b|tojson }}]] = tail[idx[{{ p.b|tojson }}]][idx[{{ p.a|tojson }}]] = {{ 'null' if p.tail is none else p.tail }};
{% endfor %}
for (var i = 0; i < edges.length; i++) { full[i][i] = 1.0; tail[i][i] = 1.0; }
Plotly.newPlot('pf-heatmap',
[{z: full, x: edges, y: edges, type: 'heatmap', colorscale: 'RdBu', zmid: 0, zmin: -1, zmax: 1,
name: 'full', colorbar: {title: 'ρ'}}],
{margin: {t: 10, r: 10, b: 60, l: 80}, paper_bgcolor: 'transparent', font: {color: '#c9d1d9'},
updatemenus: [{buttons: [
{label: 'Full', method: 'restyle', args: ['z', [full]]},
{label: 'Left-tail', method: 'restyle', args: ['z', [tail]]}], x: 0, y: 1.15}]},
{displayModeBar: false, responsive: true});
})();
</script>
{% endif %}
```
- [ ] **Step 6: Add Plotly script + nav links to base.html**
In `src/fxhnt/adapters/web/templates/base.html`, directly after the vendored htmx `<script>` line (line ~7), add:
```html
{# Plotly loads from CDN in the operator's BROWSER over the tailscale-served page — this is NOT pod egress,
so the vendored-htmx "no CDN / tailnet egress" note above does not apply here. Charting layer for
/paper/portfolio (existing server-SVG charts migrate later). SRI-pinned + crossorigin so a compromised
CDN can't inject script (the tag is refused if the bytes don't match the hash). #}
<script src="https://cdn.plot.ly/plotly-2.35.2.min.js"
integrity="sha384-<FILL-FROM-CDN>" crossorigin="anonymous" charset="utf-8"></script>
```
**SRI hash (required — do not ship a bare CDN tag).** Compute the `sha384` for the exact pinned version before adding the tag:
```bash
curl -sS https://cdn.plot.ly/plotly-2.35.2.min.js | openssl dgst -sha384 -binary | openssl base64 -A
```
Paste the result into `integrity="sha384-<...>"`. Without a matching hash the browser refuses to run the script, so this is verified the moment the chart renders (Task 4). If the operator prefers zero external dependency later, vendoring Plotly to `/static` (like `htmx.min.js`) is the drop-in alternative — out of scope for this slice, noted for the charting-migration slice.
In the desktop `.bar` nav (after the Backtest link, ~line 165):
```html
<a href="/paper/portfolio" class="{{ 'on' if path.startswith('/paper/portfolio') }}">Portfolio</a>
```
In the mobile `.tabs` nav (after the Backtest tab, ~line 173):
```html
<a href="/paper/portfolio" class="tab {{ 'on' if path.startswith('/paper/portfolio') }}"><span class="ic" aria-hidden="true">◪</span>Portfolio</a>
```
- [ ] **Step 7: Run the web tests**
Run: `pytest tests/integration/test_portfolio_web.py -v`
Expected: PASS (4 passed). If `create_app`'s kwargs differ, fix the test's construction (Step 2 note). If `display_name` or `_now` aren't in scope in the new handlers, import/reference them as the neighboring handlers do.
- [ ] **Step 8: Run the full suite to confirm nothing else broke**
Run: `pytest -q`
Expected: all green (the change is purely additive; no existing test should change).
- [ ] **Step 9: Commit**
```bash
git add src/fxhnt/adapters/web/app.py src/fxhnt/adapters/web/templates/portfolio.html \
src/fxhnt/adapters/web/templates/_portfolio_result.html \
src/fxhnt/adapters/web/templates/base.html tests/integration/test_portfolio_web.py
git commit -m "feat(portfolio): /paper/portfolio route + Plotly curve/heatmap + marginal-Sharpe table"
```
---
## Task 4: Local browser verification
**Files:** none (verification only — per `feedback_verify_cockpit_locally_not_prod`).
- [ ] **Step 1: Run the cockpit locally** against a DB that has the precomputed series (or the seeded in-memory fixture), following `reference_fxhnt_local_cockpit_dev_loop`. Open `/paper/portfolio` in a real browser.
- [ ] **Step 2: Verify by reading the numbers** (not just "it renders"):
- Select `crypto_tstrend` + `unlock` + `multistrat`, method = Book, full range → the equity curve draws, the metrics row shows plausible Sharpe/CAGR/maxDD, the heatmap shows a 3×3 with the Full/Left-tail toggle working, the marginal-Sharpe table lists all three.
- Switch to Explicit weights, set 0.7/0.3 on two edges → curve/metrics change.
- Narrow the date range → fewer days, metrics update.
- Deselect everything and run → the "select at least one edge" note, NOT a 500.
- Check mobile width (≤640px): the form + tables reflow (the `data-label` stacked-card pattern), charts stay within the viewport (no horizontal body scroll).
- [ ] **Step 3:** If any number looks wrong, STOP and use `superpowers:systematic-debugging` (root cause before fix). Do NOT patch the template to hide a bad number.
---
## Self-Review Notes (for the executor)
- **Spec coverage:** Success criteria 1–4 map to Tasks 1–3 (aggregator + book/explicit report + corr/marginal + read-only). Criterion 5 (Plotly renders over tailscale) maps to Task 4.
- **`BacktestStats` fields:** Task 2 assumes `cagr`/`sharpe`/`max_drawdown`/`vol`. The executor MUST read `src/fxhnt/domain/backtest.py` and use the real attribute names — this is the one place the plan cannot verify without the file open. This is called out explicitly in Task 2 Step 3/4.
- **`create_app` kwargs:** Task 3 assumes `create_app(repo=, bybit_paper_repo=, paper_repo=)`. The executor confirms the real signature (Task 3 Step 2 note) and adjusts the test; the `_paper_repo()` helper inside `app.py` is the source of truth for which repo the report reads.
- **Two correlation axes:** `combine_book` → epoch-day dict; `combine_returns` → date-string tuples. The bundler bridges them (Task 2). Diagnostics (correlation, marginal) always run on the epoch-day-aligned `arrs` from `align_edges`, independent of the chosen method — matching the spec ("weight-independent properties of the edges").

90
docs/superpowers/specs/2026-07-21-portfolio-backtester-design.md Normal file
View File

@@ -0,0 +1,90 @@
# Portfolio Backtester — Design Spec (2026-07-21)
**Status:** design approved (all decisions made with operator). Implementation plan follows in a fresh session.
Slice 1 of the "cockpit = the platform" sequence (agreed order: portfolio-backtester → trade register →
cards-to-tables/IA polish).
## Goal
A new cockpit surface where you assemble an arbitrary set of edges, choose a weighting method + date range, and
get portfolio-level Sharpe/CAGR/maxDD **plus** the diversification view the research actually cares about — the
pairwise correlation matrix (full AND left-tail) and each edge's marginal-Sharpe contribution. Reuses the
existing, production-tested combiner/metrics/diversification engine end-to-end; new code is glue + a page.
## Why this shape
The three-explorer map (2026-07-21) established: the portfolio-backtest ENGINE already exists and is battle-tested
(`combine_book`, `combine_returns`, `compute_stats`, `marginal_sharpe`, `full_correlation`,
`left_tail_correlation`), and per-edge daily return series are already MATERIALIZED and queryable (three tables).
So a portfolio backtester is glue + surface, not a new engine. It exposes the exact machinery the research arc
already trusts (inverse-vol book, marginal-Sharpe LOO gate, co-crash / left-tail lens) as an interactive tool.
Pure additive + read-only — cannot break the live fund.
## Decisions (all made with the operator)
1. **Both combiners, default `combine_book`.** Default = the live inverse-vol → cap → optional marginal-prune →
vol-target + Kelly method (`book_allocator.combine_book`, book_allocator.py:117) — "what does my real
allocation do with these edges?". Optional **explicit-weights what-if** mode via `combine_returns`
(domain/portfolio/combine.py:10) — "what if positioning were 40%?".
2. **Three-part report:** (a) equity curve + headline metrics (Sharpe/CAGR/maxDD/vol/worst-year) via
`compute_stats`; (b) **correlation heatmap showing BOTH full and left-tail** correlation per pair
(`full_correlation` + `left_tail_correlation`, tail_frac 0.10) — the honest co-crash lens; (c) per-edge
**marginal-Sharpe** table (leave-one-out contribution, `marginal_sharpe`) — the same accept/reject gate.
Correlation + marginal-Sharpe are computed on the RAW edge series (properties of the edges, weight-independent);
the curve + metrics reflect the chosen weighting. Both lenses shown.
3. **New page `/paper/portfolio`** — a dedicated first-class cockpit section (the first IA step toward
cockpit-as-platform), NOT an extension of `/paper/sim` (which stays "the 4 canonical books"). Reuses existing
dashboard styling + the table pattern (with `data-label` mobile reflow).
4. **Charting = Plotly via CDN.** Adopt Plotly as the new platform charting layer, loaded from CDN (operator's
choice; correctly noted the cockpit renders in the operator's browser over tailscale serve, so browser fetches
are NOT blocked by the pod NetworkPolicy — the earlier "CDN is blocked" reasoning was wrong). Accepted
trade-off: CDN is a browser-side dependency (small SPOF/telemetry cost) for a private single-user cockpit.
The existing server-SVG charts (charts.py sparklines/equity/overlay) migrate to Plotly LATER, in a dedicated
charting-migration slice — NOT in this slice (keeps slice 1 focused).
## Available edges (the multiselect options)
- **Bybit 4 sleeves** from `bybit_sleeve_ret` (paper_repo.bybit_sleeve_returns() → {sleeve:{epoch_day:ret}}):
`crypto_tstrend`, `unlock`, `xsfunding`, `positioning`.
- **IBKR recompute-replay books** from `sim_curve_ret` (paper_repo.sim_curve_returns(sid) → {epoch_day:ret}):
`multistrat` (+ `multistrat_levered`).
- (Optionally `paper_sleeve_ret` live-paper series — a later addition; slice 1 covers the two stored-backtest sources.)
## New code (glue + surface only)
1. **Cross-store series aggregator** — a function that unions the three stores into one
`series_by_edge: dict[str, dict[int,float]]` for an arbitrary selected edge set, regardless of venue. This is
the ONE missing data primitive; `combine_book`/`combine_returns` both already consume exactly this shape.
2. **Report bundler** — takes selected edges + weighting method (+ optional explicit weights) + date range →
slices the series to the range → calls `combine_book` (or `combine_returns`) for the curve, `compute_stats`
for metrics, `full_correlation`+`left_tail_correlation` for the N×N matrix, `marginal_sharpe` per edge → one
report object. Never called together on a user-defined portfolio today (only em_asia_marginal_eval does a
fixed candidate-vs-book version) — this is the new assembly.
3. **Route + template + Plotly rendering** — `GET /paper/portfolio` (page) + `GET /paper/portfolio/run` (HTMX
partial: edge-multiselect + weighting toggle + date range → the three-part report). Plotly `<script>` from
CDN in base.html (or the portfolio template). Equity curve + heatmap = Plotly; marginal-Sharpe = an HTML
table (seeds the interactive-datatable pattern the register slice will reuse).
## Reused (exists, production-tested — do NOT rebuild)
`align_edges` + `combine_book` (book_allocator.py), `combine_returns` (domain/portfolio/combine.py),
`compute_stats` (domain/backtest.py), `marginal_sharpe` (domain/diversification/marginal.py),
`full_correlation`/`left_tail_correlation` (domain/diversification/correlation.py), the stored return series
(`bybit_sleeve_returns`, `sim_curve_returns`), the existing DashboardService wiring + dark design system.
## Out of scope (this slice)
- Migrating the existing server-SVG charts to Plotly (own slice later).
- Bybit/IBKR real-fill ingestion, the trade register, the EUR ledger (slice 2).
- Live-paper `paper_sleeve_ret` as a selectable source (a small later add).
- Persisting/saving named portfolios (could add; not needed for v1).
## Success criteria
1. Select any subset of the available edges + a date range → get a portfolio equity curve + Sharpe/CAGR/maxDD
in the cockpit, using `combine_book` by default, instantly (series are pre-materialized — no recompute).
2. Explicit-weights mode lets you supply a weight per selected edge → curve/metrics reflect those weights.
3. The correlation heatmap shows full AND left-tail correlation per pair; the marginal-Sharpe table shows each
edge's LOO contribution — matching the numbers the research pipeline uses to accept/reject edges.
4. Read-only + additive: no live fund path touched; `/paper/sim` and everything else unchanged.
5. Plotly charts render over the tailscale-served cockpit in the browser.

57
src/fxhnt/adapters/web/app.py
View File

@@ -2,8 +2,10 @@
takes the repo so tests can inject a seeded in-memory DB; create_app() builds it from settings."""
from __future__ import annotations
import contextlib
import datetime as dt
import logging
import math
import time
from pathlib import Path
from typing import TYPE_CHECKING, Any
@@ -764,6 +766,61 @@ def create_app(repo: ForwardNavRepo, svc: DashboardReadModel | None = None,
mctx = _measured_pending_context(book)
return templates.TemplateResponse(request, "_sim_result.html", mctx)
@app.get("/paper/portfolio", response_class=HTMLResponse)
def paper_portfolio(request: Request) -> HTMLResponse:
"""Render the edge-multiselect + method toggle + date-range form INSTANTLY; the three-part report
(curve + full/left-tail correlation heatmap + marginal-Sharpe table) lazy-loads from
/paper/portfolio/run. Read-only: reads ONLY the precomputed stored series (no edge recompute)."""
from fxhnt.application.portfolio_backtest import BYBIT_EDGES, SIM_EDGES
# Pre-fill the date bounds from the union range of the available stored series so the pickers are
# never blank and can't go off-data.
repo = _paper_repo()
bybit = {}
try:
bybit = repo.bybit_sleeve_returns()
except Exception: # noqa: BLE001 — a missing table must never 500 the form
bybit = {}
all_days: set[int] = set()
for s in bybit.values():
all_days |= set(s)
with contextlib.suppress(Exception): # a missing table must never 500 the form
all_days |= set(repo.sim_curve_returns("multistrat"))
lo = dt.date.fromordinal(min(all_days)).isoformat() if all_days else ""
hi = dt.date.fromordinal(max(all_days)).isoformat() if all_days else ""
ctx = {"bybit_edges": BYBIT_EDGES, "sim_edges": SIM_EDGES,
"range_min": lo, "range_max": hi, "now": _now(),
"labels": {e: display_name(e) for e in (*BYBIT_EDGES, *SIM_EDGES)}}
return templates.TemplateResponse(request, "portfolio.html", ctx)
@app.get("/paper/portfolio/run", response_class=HTMLResponse)
def paper_portfolio_run(request: Request) -> HTMLResponse:
"""Build the report for the selected edges + method + range and render the three-part partial.
Empty selection / missing series degrade to a 'select an edge' / 'precomputing' note, never a 500."""
from fxhnt.application.portfolio_backtest import ALL_EDGES, build_portfolio_report
qp = request.query_params
edges = [e for e in qp.getlist("edges") if e in ALL_EDGES]
method = qp.get("method") if qp.get("method") in ("book", "explicit") else "book"
start = qp.get("start") or None
end = qp.get("end") or None
# Parse per-edge weights defensively: only FINITE, NON-NEGATIVE values are accepted (a query like
# w_x=inf / w_x=nan / w_x=-0.5 must not reach the math and produce an Infinity/NaN/negative-weighted
# curve — the number inputs' min=0 is client-side only, so the server is the real guard).
weights: dict[str, float] = {}
for e in edges:
raw = qp.get(f"w_{e}")
if raw:
with contextlib.suppress(ValueError):
v = float(raw)
if math.isfinite(v) and v >= 0.0:
weights[e] = v
try:
report = build_portfolio_report(_paper_repo(), edges=edges, method=method,
weights=weights or None, start=start, end=end)
except Exception: # noqa: BLE001 — degrade to an empty report, never 500
report = {"empty": True, "edges": [], "method": method, "dates_iso": [], "equity": [],
"returns": [], "metrics": {}, "correlation": [], "marginal": []}
return templates.TemplateResponse(request, "_portfolio_result.html", {"r": report})
def _venue_repo() -> PaperRepo:
"""The Bybit PaperRepo — the cockpit's only paper-book venue (the injected one, or one built from
settings). A LIGHT read-model store; used only for the persisted nav-history read, no

8
src/fxhnt/adapters/web/static/plotly.min.js vendored Normal file
View File

File diff suppressed because one or more lines are too long

63
src/fxhnt/adapters/web/templates/_portfolio_result.html Normal file
View File

@@ -0,0 +1,63 @@
{# src/fxhnt/adapters/web/templates/_portfolio_result.html #}
{% if r.empty %}
<p class="muted">Select at least one edge and run the backtest.</p>
{% else %}
{# The ACTIVE weighting method — the bundler relabels method to "book" when "explicit" was asked for but
no usable weights were supplied, so showing it here makes that silent fallback visible to the operator. #}
<p class="muted">Weighting: <strong>{{ 'Explicit weights' if r.method == 'explicit' else 'Book (live allocator)' }}</strong></p>
<div class="metrics-row">
<div class="metric"><span class="k">CAGR</span><span class="v">{{ '%.1f'|format(r.metrics.cagr * 100) }}%</span></div>
<div class="metric"><span class="k">Sharpe</span><span class="v">{{ '%.2f'|format(r.metrics.sharpe) }}</span></div>
<div class="metric"><span class="k">Max DD</span><span class="v">{{ '%.1f'|format(r.metrics.max_drawdown * 100) }}%</span></div>
<div class="metric"><span class="k">Vol</span><span class="v">{{ '%.1f'|format(r.metrics.vol * 100) }}%</span></div>
<div class="metric"><span class="k">Days</span><span class="v">{{ r.metrics.n_days }}</span></div>
</div>
<div id="pf-equity" class="chart"></div>
<h3>Correlation (full / left-tail)</h3>
<div id="pf-heatmap" class="chart"></div>
<h3>Marginal Sharpe (leave-one-out)</h3>
<table class="data">
<thead><tr><th>Edge</th><th>Marginal Sharpe</th></tr></thead>
<tbody>
{% for m in r.marginal %}
<tr><td data-label="Edge">{{ m.edge }}</td>
<td data-label="Marginal Sharpe">{{ '—' if m.marginal_sharpe is none else '%.3f'|format(m.marginal_sharpe) }}</td></tr>
{% endfor %}
</tbody>
</table>
{# Two server-rendered matrices (full + left-tail) → one heatmap with a Full/Left-tail toggle. Jinja renders
the matrices server-side (a JS closure can't parameterize a Jinja loop), so the JS is pure Plotly. #}
<script>
(function () {
var dates = {{ r.dates_iso|tojson }};
var equity = {{ r.equity|tojson }};
var edges = {{ r.marginal|map(attribute='edge')|list|tojson }};
Plotly.newPlot('pf-equity',
[{x: dates, y: equity, type: 'scatter', mode: 'lines', line: {color: '#4c9aff'}}],
{margin: {t: 10, r: 10, b: 30, l: 40}, paper_bgcolor: 'transparent',
plot_bgcolor: 'transparent', font: {color: '#c9d1d9'}, yaxis: {title: 'Equity (×)'}},
{displayModeBar: false, responsive: true});
var idx = {}; edges.forEach(function (e, i) { idx[e] = i; });
function blank() { return edges.map(function () { return edges.map(function () { return null; }); }); }
var full = blank(), tail = blank();
{% for p in r.correlation %}
full[idx[{{ p.a|tojson }}]][idx[{{ p.b|tojson }}]] = full[idx[{{ p.b|tojson }}]][idx[{{ p.a|tojson }}]] = {{ 'null' if p.full is none else p.full }};
tail[idx[{{ p.a|tojson }}]][idx[{{ p.b|tojson }}]] = tail[idx[{{ p.b|tojson }}]][idx[{{ p.a|tojson }}]] = {{ 'null' if p.tail is none else p.tail }};
{% endfor %}
for (var i = 0; i < edges.length; i++) { full[i][i] = 1.0; tail[i][i] = 1.0; }
Plotly.newPlot('pf-heatmap',
[{z: full, x: edges, y: edges, type: 'heatmap', colorscale: 'RdBu', zmid: 0, zmin: -1, zmax: 1,
name: 'full', colorbar: {title: 'ρ'}}],
{margin: {t: 10, r: 10, b: 60, l: 80}, paper_bgcolor: 'transparent', font: {color: '#c9d1d9'},
updatemenus: [{buttons: [
{label: 'Full', method: 'restyle', args: ['z', [full]]},
{label: 'Left-tail', method: 'restyle', args: ['z', [tail]]}], x: 0, y: 1.15}]},
{displayModeBar: false, responsive: true});
})();
</script>
{% endif %}

5
src/fxhnt/adapters/web/templates/base.html
View File

@@ -5,6 +5,7 @@
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{% block title %}cockpit{% endblock %} · fxhnt</title>
<script src="/static/htmx.min.js"></script> <!-- vendored, no CDN: works on the tailnet w/o external egress -->
<script src="/static/plotly.min.js"></script> <!-- vendored, no CDN: leak-clean (relative URL only) + no external egress on the tailnet, same as htmx.min.js above. Charting layer for /paper/portfolio. -->
<style>
/* ── Design system: a professional trading-app dark theme. ONE palette (CSS vars), sans for prose +
@@ -126,7 +127,7 @@
nav.bar { display:none; } /* hide the top nav row on mobile */
main { padding:12px 12px 76px 12px; } /* clear the fixed bottom bar */
nav.tabs { position:fixed; bottom:0; left:0; right:0; z-index:20; display:grid;
grid-template-columns:repeat(4,1fr); background:rgba(13,17,23,.94);
grid-template-columns:repeat(5,1fr); background:rgba(13,17,23,.94);
backdrop-filter:blur(10px); border-top:1px solid #21262d; }
nav.tabs .tab { padding:9px 0 11px; text-align:center; color:#6e7681; font-size:10.5px;
letter-spacing:.02em; }
@@ -163,6 +164,7 @@
<a href="/paper" class="{{ 'on' if path == '/paper' or path.startswith('/paper/crypto') }}">Paper</a>
<a href="/paper/replay" class="{{ 'on' if path.startswith('/paper/replay') }}">Replay</a>
<a href="/paper/sim" class="{{ 'on' if path.startswith('/paper/sim') }}">Backtest</a>
<a href="/paper/portfolio" class="{{ 'on' if path.startswith('/paper/portfolio') }}">Portfolio</a>
</nav>
<span class="muted" style="margin-left:auto">{{ now }}</span></header>
{# mobile bottom-tab nav: thumb-reachable, shown only <=640px (see CSS). Mirrors the desktop nav.bar. #}
@@ -171,6 +173,7 @@
<a href="/paper" class="tab {{ 'on' if path == '/paper' or path.startswith('/paper/crypto') }}"><span class="ic" aria-hidden="true">▤</span>Paper</a>
<a href="/paper/replay" class="tab {{ 'on' if path.startswith('/paper/replay') }}"><span class="ic" aria-hidden="true">▶</span>Replay</a>
<a href="/paper/sim" class="tab {{ 'on' if path.startswith('/paper/sim') }}"><span class="ic" aria-hidden="true">◷</span>Backtest</a>
<a href="/paper/portfolio" class="tab {{ 'on' if path.startswith('/paper/portfolio') }}"><span class="ic" aria-hidden="true">◪</span>Portfolio</a>
</nav>
<main>{% block body %}{% endblock %}</main>
</body>

53
src/fxhnt/adapters/web/templates/portfolio.html Normal file
View File

@@ -0,0 +1,53 @@
{# src/fxhnt/adapters/web/templates/portfolio.html #}
{% extends "base.html" %}
{% block title %}Portfolio{% endblock %}
{% block body %}
<section class="card">
<h2>Portfolio Backtest</h2>
<p class="muted">Assemble edges, pick a weighting method + date range, and see the portfolio-level
Sharpe/CAGR/maxDD plus the correlation (full &amp; left-tail) and marginal-Sharpe diagnostics —
all from the precomputed series (no recompute).</p>
<form hx-get="/paper/portfolio/run" hx-target="#portfolio-result" hx-trigger="submit"
hx-indicator="#portfolio-loading">
<fieldset>
<legend>Edges</legend>
{% for e in bybit_edges %}
<label class="chk"><input type="checkbox" name="edges" value="{{ e }}"> {{ labels[e] }}</label>
{% endfor %}
{% for e in sim_edges %}
<label class="chk"><input type="checkbox" name="edges" value="{{ e }}"> {{ labels[e] }}</label>
{% endfor %}
</fieldset>
<fieldset>
<legend>Method</legend>
<label class="chk"><input type="radio" name="method" value="book" checked> Book (live allocator)</label>
<label class="chk"><input type="radio" name="method" value="explicit"> Explicit weights</label>
</fieldset>
<fieldset id="explicit-weights">
<legend>Explicit weights (used only in “explicit” mode)</legend>
{% for e in bybit_edges %}
<label>{{ labels[e] }} <input type="number" step="0.05" min="0" max="1" name="w_{{ e }}"></label>
{% endfor %}
{% for e in sim_edges %}
<label>{{ labels[e] }} <input type="number" step="0.05" min="0" max="1" name="w_{{ e }}"></label>
{% endfor %}
</fieldset>
<fieldset>
<legend>Date range</legend>
<label>Start <input type="date" name="start" value="{{ range_min }}"
min="{{ range_min }}" max="{{ range_max }}"></label>
<label>End <input type="date" name="end" value="{{ range_max }}"
min="{{ range_min }}" max="{{ range_max }}"></label>
</fieldset>
<button type="submit">Run backtest</button>
<span id="portfolio-loading" class="htmx-indicator muted">running…</span>
</form>
<div id="portfolio-result"></div>
</section>
{% endblock %}

155
src/fxhnt/application/portfolio_backtest.py Normal file
View File

@@ -0,0 +1,155 @@
"""Portfolio backtester (cockpit slice 1): assemble an arbitrary set of edges from the PRECOMPUTED stored
return series, run them through the EXISTING book combiner / metrics / diversification engine, and return one
report the /paper/portfolio surface renders. Pure application logic — reads only the two repo accessors, never
recomputes an edge/replay on the request path. Read-only; cannot touch the live fund."""
from __future__ import annotations
import datetime as dt
from itertools import combinations
from typing import Protocol
import numpy as np
from fxhnt.application.book_allocator import align_edges, combine_book
from fxhnt.domain.backtest import compute_stats
from fxhnt.domain.diversification.correlation import (
full_correlation,
left_tail_correlation,
)
from fxhnt.domain.diversification.marginal import marginal_sharpe
from fxhnt.domain.portfolio.combine import combine_returns
# The selectable edges, grouped by their stored-series source (v1 fixed set).
BYBIT_EDGES: tuple[str, ...] = ("crypto_tstrend", "unlock", "xsfunding", "positioning")
SIM_EDGES: tuple[str, ...] = ("multistrat", "multistrat_levered")
ALL_EDGES: tuple[str, ...] = BYBIT_EDGES + SIM_EDGES
class _SeriesRepo(Protocol):
def bybit_sleeve_returns(self) -> dict[str, dict[int, float]]: ...
def sim_curve_returns(self, strategy_id: str) -> dict[int, float]: ...
def aggregate_edge_series(repo: _SeriesRepo, edges: list[str]) -> dict[str, dict[int, float]]:
"""Union the selected edges' PRECOMPUTED per-day series into one `{edge: {epoch_day: ret}}` map — the
exact shape `book_allocator.align_edges`/`combine_book` consume. Bybit sleeves come from the one
`bybit_sleeve_returns()` read; each IBKR recompute-replay book from its own `sim_curve_returns(id)` read.
Unknown edges and edges whose stored series is empty are silently dropped (a selection referencing a
not-yet-precomputed book must degrade, never error)."""
bybit = repo.bybit_sleeve_returns()
out: dict[str, dict[int, float]] = {}
for e in edges:
if e in BYBIT_EDGES:
series = bybit.get(e) or {}
elif e in SIM_EDGES:
series = repo.sim_curve_returns(e)
else:
continue # unknown edge name — drop
if series:
out[e] = dict(series)
return out
_PERIODS_PER_YEAR = 365 # stored series are calendar-daily (matches combine_book's default)
def _window(series_by_edge: dict[str, dict[int, float]], start: str | None, end: str | None
) -> dict[str, dict[int, float]]:
"""Clip every edge's {epoch_day: ret} to the inclusive [start, end] ISO date range (either bound optional)."""
lo = dt.date.fromisoformat(start).toordinal() if start else None
hi = dt.date.fromisoformat(end).toordinal() if end else None
out: dict[str, dict[int, float]] = {}
for e, s in series_by_edge.items():
clipped = {d: r for d, r in s.items()
if (lo is None or d >= lo) and (hi is None or d <= hi)}
if clipped:
out[e] = clipped
return out
def _empty_report(method: str) -> dict:
return {"edges": [], "method": method, "dates_iso": [], "equity": [], "returns": [],
"metrics": {"cagr": 0.0, "sharpe": 0.0, "max_drawdown": 0.0, "vol": 0.0, "n_days": 0},
"correlation": [], "marginal": [], "empty": True}
def _metrics_from_stats(returns: np.ndarray) -> dict:
if returns.size == 0:
return {"cagr": 0.0, "sharpe": 0.0, "max_drawdown": 0.0, "vol": 0.0, "n_days": 0}
st = compute_stats(returns, periods_per_year=_PERIODS_PER_YEAR)
# NOTE: BacktestStats (src/fxhnt/domain/models.py) exposes `ann_vol`, not `vol` — mapped below.
return {"cagr": float(st.cagr), "sharpe": float(st.sharpe),
"max_drawdown": float(st.max_drawdown), "vol": float(st.ann_vol), "n_days": int(returns.size)}
def _equity_curve(returns: np.ndarray) -> list[float]:
"""Compound daily returns from a $1.0 base (the surface scales by capital client-side / in the template)."""
eq, curve = 1.0, []
for r in returns:
eq *= (1.0 + float(r))
curve.append(eq)
return curve
def _nan_to_none(x: float) -> float | None:
"""JSON/template-safe: the domain fns return NaN when a series is too short to estimate; surface as None."""
return None if x is None or (isinstance(x, float) and np.isnan(x)) else float(x)
def build_portfolio_report(repo: _SeriesRepo, *, edges: list[str], method: str = "book",
weights: dict[str, float] | None = None,
start: str | None = None, end: str | None = None) -> dict:
"""Assemble the full portfolio report for the selected edges + weighting method + date range.
method="book" -> the live inverse-vol/cap/vol-target+Kelly allocator (`combine_book`) — "what does my
real allocation do with these edges?".
method="explicit" -> a plain capital-weighted blend of the supplied per-edge `weights` (`combine_returns`) —
"what if positioning were 40%?".
The equity/metrics reflect the chosen weighting; the correlation matrix (full AND left-tail) and per-edge
marginal-Sharpe are computed on the RAW aligned edge arrays (properties of the edges, weight-independent).
Read-only; never recomputes an edge."""
series = _window(aggregate_edge_series(repo, edges), start, end)
if not series:
return _empty_report(method)
# RAW aligned per-edge arrays on the shared epoch-day axis (0.0-fill on flat days) — the basis for the
# book curve's diagnostics (correlation + marginal-Sharpe).
epoch_days, arrs = align_edges(series)
if not epoch_days:
return _empty_report(method)
used = list(arrs)
if method == "explicit" and weights:
# combine_returns aligns on the INTERSECTION of date-STRING axes; build one Sleeve per used edge.
iso = [dt.date.fromordinal(d).isoformat() for d in epoch_days]
sleeves = [(float(weights.get(e, 0.0)), tuple(iso), arrs[e]) for e in used]
_cdates, book_ret = combine_returns(sleeves)
# combine_returns may drop to the intersection; re-derive the ISO axis from its returned dates.
dates_iso = list(_cdates)
returns = np.asarray(book_ret, dtype=float)
else:
method = "book"
book_map = combine_book(series) # {epoch_day: book_return}
book_days = sorted(book_map)
dates_iso = [dt.date.fromordinal(d).isoformat() for d in book_days]
returns = np.array([book_map[d] for d in book_days], dtype=float)
if returns.size == 0:
return _empty_report(method)
# Diagnostics on the raw edge arrays — pairwise, weight-independent (aligned to min length inside
# the domain fns). `tail` is the pair's own left-tail co-movement (edge-a's tail vs edge-b), not the
# book's tail — a true per-pair number for the heatmap.
correlation = [{"a": a, "b": b,
"full": _nan_to_none(full_correlation(arrs[a], arrs[b])),
"tail": _nan_to_none(left_tail_correlation(arrs[a], arrs[b]))}
for a, b in combinations(used, 2)]
book_arr = returns
marginal = [{"edge": e, "marginal_sharpe": _nan_to_none(marginal_sharpe(book_arr, arrs[e]))}
for e in used]
return {"edges": used, "method": method, "dates_iso": dates_iso,
"equity": _equity_curve(returns), "returns": [float(r) for r in returns],
"metrics": _metrics_from_stats(returns), "correlation": correlation,
"marginal": marginal, "empty": False}

113
tests/application/test_portfolio_backtest.py Normal file
View File

@@ -0,0 +1,113 @@
"""Unit tests for the portfolio-backtest aggregator + report bundler (read-only, pure)."""
from __future__ import annotations
import datetime as dt
from fxhnt.application.portfolio_backtest import (
aggregate_edge_series,
build_portfolio_report,
)
class _FakeRepo:
"""Minimal stand-in exposing only the two accessors the aggregator reads."""
def __init__(self, bybit: dict[str, dict[int, float]], sim: dict[str, dict[int, float]]):
self._bybit = bybit
self._sim = sim
def bybit_sleeve_returns(self) -> dict[str, dict[int, float]]:
return self._bybit
def sim_curve_returns(self, strategy_id: str) -> dict[int, float]:
return self._sim.get(strategy_id, {})
def _d(y: int, m: int, day: int) -> int:
return dt.date(y, m, day).toordinal()
def test_aggregate_unions_bybit_and_sim_sources():
repo = _FakeRepo(
bybit={
"crypto_tstrend": {_d(2026, 1, 1): 0.01, _d(2026, 1, 2): -0.02},
"unlock": {_d(2026, 1, 1): 0.003},
},
sim={"multistrat": {_d(2026, 1, 1): 0.005, _d(2026, 1, 2): 0.004}},
)
out = aggregate_edge_series(repo, ["crypto_tstrend", "multistrat"])
assert set(out) == {"crypto_tstrend", "multistrat"}
assert out["crypto_tstrend"][_d(2026, 1, 2)] == -0.02
assert out["multistrat"][_d(2026, 1, 1)] == 0.005
def test_aggregate_skips_unknown_and_empty_edges():
repo = _FakeRepo(bybit={"unlock": {_d(2026, 1, 1): 0.003}}, sim={})
out = aggregate_edge_series(repo, ["unlock", "does_not_exist", "multistrat"])
assert set(out) == {"unlock"} # unknown + empty-series edges dropped
def _series(start_ord: int, rets: list[float]) -> dict[int, float]:
return {start_ord + i: r for i, r in enumerate(rets)}
def _repo_with_overlap() -> _FakeRepo:
s = _d(2026, 1, 1)
# 40 days so trailing windows in combine_book/left-tail have room.
a = [0.01 if i % 2 else -0.008 for i in range(40)]
b = [0.006 if i % 3 else -0.004 for i in range(40)]
c = [0.004 for _ in range(40)]
return _FakeRepo(
bybit={"crypto_tstrend": _series(s, a), "unlock": _series(s, b)},
sim={"multistrat": _series(s, c)},
)
def test_report_book_method_has_curve_metrics_corr_marginal():
rep = build_portfolio_report(
_repo_with_overlap(), edges=["crypto_tstrend", "unlock", "multistrat"], method="book"
)
assert rep["empty"] is False
assert rep["method"] == "book"
assert set(rep["edges"]) == {"crypto_tstrend", "unlock", "multistrat"}
assert len(rep["dates_iso"]) == len(rep["equity"]) == len(rep["returns"])
assert rep["equity"][0] > 0.0 # compounded from a positive base
for k in ("cagr", "sharpe", "max_drawdown", "vol", "n_days"):
assert k in rep["metrics"]
# 3 edges -> 3 unordered pairs.
assert len(rep["correlation"]) == 3
for pair in rep["correlation"]:
assert set(pair) == {"a", "b", "full", "tail"}
assert len(rep["marginal"]) == 3
assert all(set(m) == {"edge", "marginal_sharpe"} for m in rep["marginal"])
def test_report_explicit_weights_method_uses_supplied_weights():
rep = build_portfolio_report(
_repo_with_overlap(),
edges=["crypto_tstrend", "unlock"],
method="explicit",
weights={"crypto_tstrend": 0.7, "unlock": 0.3},
)
assert rep["empty"] is False
assert rep["method"] == "explicit"
assert len(rep["returns"]) > 0
def test_report_empty_when_no_edges_selected():
rep = build_portfolio_report(_repo_with_overlap(), edges=[], method="book")
assert rep["empty"] is True
assert rep["equity"] == []
assert rep["correlation"] == []
def test_report_date_range_windows_the_series():
repo = _repo_with_overlap()
full = build_portfolio_report(repo, edges=["crypto_tstrend", "unlock"], method="book")
windowed = build_portfolio_report(
repo, edges=["crypto_tstrend", "unlock"], method="book",
start="2026-01-10", end="2026-01-20",
)
assert len(windowed["dates_iso"]) < len(full["dates_iso"])
assert windowed["dates_iso"][0] >= "2026-01-10"
assert windowed["dates_iso"][-1] <= "2026-01-20"

67
tests/integration/test_portfolio_web.py Normal file
View File

@@ -0,0 +1,67 @@
"""Web smoke for the portfolio backtester (/paper/portfolio): assemble edges from the PRECOMPUTED stored
series and render a portfolio curve + metrics + full/left-tail correlation heatmap + marginal-Sharpe table,
WITHOUT recomputing any edge on the request path. Read-only + additive: no live-fund path touched.
NO network — in-memory SQLite repos, seeded directly."""
from __future__ import annotations
import datetime as dt
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
def _seed(repo: PaperRepo) -> None:
at = dt.datetime(2026, 1, 1, tzinfo=dt.UTC)
start = dt.date(2026, 1, 1)
for i in range(40):
d = (start + dt.timedelta(days=i)).isoformat()
repo.upsert_bybit_sleeve_ret(d, "crypto_tstrend", 0.01 if i % 2 else -0.008, at=at)
repo.upsert_bybit_sleeve_ret(d, "unlock", 0.006 if i % 3 else -0.004, at=at)
repo.upsert_sim_curve_ret("multistrat", d, 0.004, at=at)
def _client() -> TestClient:
paper = PaperRepo("sqlite://")
paper.migrate()
_seed(paper)
fwd = ForwardNavRepo("sqlite://")
fwd.migrate()
app = create_app(repo=fwd, bybit_paper_repo=paper, paper_repo=paper)
return TestClient(app)
def test_portfolio_page_renders_form():
r = _client().get("/paper/portfolio")
assert r.status_code == 200
body = r.text
assert "crypto_tstrend" in body and "multistrat" in body
assert "/paper/portfolio/run" in body # the HTMX target
def test_portfolio_run_book_method_returns_curve_metrics_corr_marginal():
r = _client().get("/paper/portfolio/run",
params=[("edges", "crypto_tstrend"), ("edges", "unlock"),
("edges", "multistrat"), ("method", "book")])
assert r.status_code == 200
body = r.text
assert "Plotly" in body # the curve/heatmap render scripts
assert "Sharpe" in body # headline metrics table
assert "Marginal" in body # marginal-Sharpe table heading
def test_portfolio_run_empty_selection_degrades_gracefully():
r = _client().get("/paper/portfolio/run", params={"method": "book"})
assert r.status_code == 200 # never a 500
assert "Select at least one edge" in r.text or "no edges" in r.text.lower()
def test_portfolio_run_explicit_weights_mode():
r = _client().get("/paper/portfolio/run",
params=[("edges", "crypto_tstrend"), ("edges", "unlock"),
("method", "explicit"), ("w_crypto_tstrend", "0.7"),
("w_unlock", "0.3")])
assert r.status_code == 200
assert "Plotly" in r.text