# Paper-Snapshot Job — Implementation Plan (Phase 2F-2)

> **For agentic workers:** REQUIRED SUB-SKILL: superpowers:subagent-driven-development (recommended) or
> executing-plans. TDD throughout. Steps use `- [ ]`.

**Goal:** Populate the `/paper` cockpit with the live sleeves' real positions: each sleeve exposes
`current_weights(as_of)`, a `PaperSnapshotService` combines them with allocator sleeve-weights + warehouse
closes → `persist_paper_book`, wired as a nightly Dagster asset.

**Architecture:** Read-only/additive — a parallel `current_weights()` accessor per sleeve (reusing its
existing domain weight logic) + a snapshot service + a Dagster asset after `combined_forward_nav`. The
return-producing hot path + gate record are untouched.

**Tech Stack:** Python 3.12, Dagster, SQLAlchemy, pytest. **Repo:** `~/Work/fxhnt`, branch
`feat/paper-snapshot`. **Spec:** `docs/superpowers/specs/2026-06-21-paper-snapshot-design.md`.
**Run tests:** `uv run pytest <path> -q`.

**Backend already available (2F MVP, merged):** `persist_paper_book(repo, *, capital, run_date,
sleeve_weights, symbol_weights_by_sleeve, prices, at, enabled) -> int` in `application/paper_book.py`;
`PaperRepo`; domain `paper.py`.

**Live sleeves (registry):** crypto-momentum (vestigial trend), 60/40 (beta), xsfunding (crypto-funding),
unlock (crypto-event), crypto ts-trend (crypto-trend), stablecoin (crypto-stablecoin).

---

## Task 1: `PositionSource` protocol

**Files:** Create `src/fxhnt/ports/position_source.py`; Test `tests/unit/test_position_source.py`

- [ ] **Step 1: Failing test**
```python
# tests/unit/test_position_source.py
from fxhnt.ports.position_source import PositionSource
class _Stub:
    def current_weights(self, as_of: str) -> dict[str, float]:
        return {"USDTUSD": -1.0}
def test_protocol_structural():
    s: PositionSource = _Stub()
    assert s.current_weights("2026-06-21") == {"USDTUSD": -1.0}
```
- [ ] **Step 2:** run → FAIL.
- [ ] **Step 3:** `ports/position_source.py`:
```python
from __future__ import annotations
from typing import Protocol
class PositionSource(Protocol):
    """A live sleeve's current per-symbol target weights (signed) as of a rebalance date.
    Returns {} when the sleeve has no data / can't surface weights (caller skips + logs)."""
    def current_weights(self, as_of: str) -> dict[str, float]: ...
```
- [ ] **Step 4:** PASS. **Step 5:** `git commit -m "feat(paper-snap): PositionSource protocol"`

---

## Task 2: stablecoin `current_weights()` (worked example — follow this pattern for the others)

**Files:** Modify `src/fxhnt/application/stablecoin_runner.py` (or the stablecoin sleeve service);
Test `tests/unit/test_stablecoin_current_weights.py`

- [ ] **Step 1: Failing test**
```python
# tests/unit/test_stablecoin_current_weights.py
from fxhnt.application.stablecoin_runner import StableReversionRunner
def test_current_weights_shorts_rich_names():
    # panel = {symbol: {epoch_day: close}}; day 20000 has USDT rich (>1+thresh), DAI at peg
    panel = {"USDTUSD": {20000: 1.01}, "DAIUSD": {20000: 1.0001}}
    r = StableReversionRunner(panel, thresh_bp=50.0)
    w = r.current_weights("epoch:20000")   # or the runner's date convention
    assert w["USDTUSD"] < 0      # rich -> short
    assert w.get("DAIUSD", 0.0) == 0.0
```
- [ ] **Step 2:** run → FAIL.
- [ ] **Step 3:** Add `current_weights(self, as_of)` to `StableReversionRunner` reusing its existing logic:
  compute `dev = {s: series[day] - 1.0 for s, series in self._p.items() if day in series}` for the latest
  available day ≤ as_of, then `return short_rich_weights(dev, self._th)`. (Mirror exactly how `run()`'s loop
  builds `dev` + calls `short_rich_weights`; reuse `self._th`.) Resolve `as_of`→epoch-day with the runner's
  existing date helper.
- [ ] **Step 4:** PASS. **Step 5:** `git commit -m "feat(paper-snap): stablecoin current_weights"`

---

## Task 3: `current_weights()` for the other crypto sleeves (unlock, xsfunding, ts-trend)

**Files:** Modify the unlock / xsfunding / ts-trend sleeve runners/services; Tests per sleeve under
`tests/unit/test_<sleeve>_current_weights.py`

For EACH of unlock, xsfunding, crypto ts-trend: add `current_weights(as_of)` reusing that sleeve's existing
domain weight fn for the latest day, with a focused unit test (reuse the sleeve's existing domain test
fixtures). Specifics (study each runner first):
- **unlock shorts** → `inverse_vol_capped_weights(vols, max_name_weight)` on the names with an active
  unlock window as of `as_of`, **negated** (shorts). Test: an in-window name gets a negative weight.
- **xsfunding** → `construction_weights(scores, mode, quantile=…)` on the latest funding scores. Test: top/
  bottom quantile get +/− weights, middle 0.
- **crypto ts-trend** → per symbol `ts_trend_signal(closes[sym], lookbacks)`; long/flat sign → weight
  (equal-weight across symbols with a positive signal, scaled to the sleeve's gross). Test: an up-trending
  series → positive weight, a down/flat one → 0.

- [ ] **Step per sleeve:** failing test → implement `current_weights` (reuse domain fn) → PASS → commit
  `feat(paper-snap): <sleeve> current_weights`.

---

## Task 4: `current_weights()` for the equity/beta sleeves (60/40, crypto-momentum)

**Files:** Modify the sixtyforty + crypto-momentum sleeve services; Tests under `tests/unit/`

- **60/40** → constant weights `{"SPY": 0.6, "IEF": 0.4}` (or the sleeve's `_SIXTYFORTY_WEIGHTS`). Test:
  returns those two weights summing to 1.0.
- **crypto-momentum (vestigial trend overlay)** → its current long/flat weight from its existing signal; if
  the sleeve is purely a vestigial overlay with no clean per-symbol weight, `current_weights` returns `{}`
  (documented — it'll be skipped). Test: returns `{}` or the documented weight.

- [ ] **Step per sleeve:** failing test → implement → PASS → commit `feat(paper-snap): <sleeve> current_weights`.

---

## Task 5: allocator `current_sleeve_weights()` accessor

**Files:** Modify `src/fxhnt/application/book_allocator.py`; Test `tests/unit/test_allocator_sleeve_weights.py`

- [ ] **Step 1: Failing test**
```python
# tests/unit/test_allocator_sleeve_weights.py
import numpy as np
from fxhnt.application.book_allocator import current_sleeve_weights
def test_current_sleeve_weights_inverse_vol():
    # two edges, edge B 2x the vol of A -> A gets more weight; weights sum ~1
    series = {"A": {1:0.001,2:-0.001,3:0.001}, "B": {1:0.01,2:-0.01,3:0.01}}
    w = current_sleeve_weights(series, vol_lookback=3, max_sleeve_weight=0.5)
    assert abs(sum(w.values()) - 1.0) < 1e-6
    assert w["A"] > w["B"]
```
- [ ] **Step 2:** run → FAIL.
- [ ] **Step 3:** Add a public `current_sleeve_weights(series_by_edge, *, vol_lookback=60,
  max_sleeve_weight=...) -> dict[str, float]` thin wrapper that runs the SAME trailing-window inverse-vol
  computation `combine_book` uses for the latest window (reuse `align_edges` + `_inverse_vol_weights` over
  the last `vol_lookback` rows). Do not change `combine_book`.
- [ ] **Step 4:** PASS. **Step 5:** `git commit -m "feat(paper-snap): allocator current_sleeve_weights"`

---

## Task 6: `PaperSnapshotService`

**Files:** Create `src/fxhnt/application/paper_snapshot.py`; Test `tests/integration/test_paper_snapshot.py`

- [ ] **Step 1: Failing test**
```python
# tests/integration/test_paper_snapshot.py
import datetime as dt
from fxhnt.adapters.persistence.paper_repo import PaperRepo
from fxhnt.application.paper_snapshot import PaperSnapshotService

class _Sleeve:
    def __init__(self, w): self._w = w
    def current_weights(self, as_of): return self._w

def test_snapshot_persists_multi_sleeve_and_skips_failures():
    repo = PaperRepo("sqlite://"); repo.migrate()
    class _Boom:
        def current_weights(self, as_of): raise RuntimeError("no data")
    svc = PaperSnapshotService(
        repo, capital=100_000.0,
        sleeves={"stablecoin_rotation": _Sleeve({"USDTUSD": -1.0}), "broken": _Boom()},
        sleeve_weights={"stablecoin_rotation": 0.6, "broken": 0.4},
        prices={"USDTUSD": 2.0})
    n = svc.snapshot(as_of="2026-06-21", at=dt.datetime(2026,6,21,tzinfo=dt.UTC))
    pos = repo.latest_positions()
    assert any(p.symbol == "USDTUSD" for p in pos)   # good sleeve persisted
    assert all(p.sleeve != "broken" for p in pos)    # failing sleeve skipped, not fatal
```
- [ ] **Step 2:** run → FAIL.
- [ ] **Step 3:** Implement `PaperSnapshotService(repo, *, capital, sleeves: dict[str, PositionSource],
  sleeve_weights: dict[str,float], prices: dict[str,float])`. `snapshot(as_of, at)`: build
  `symbol_weights_by_sleeve` by calling each sleeve's `current_weights(as_of)` inside try/except (log +
  skip on error, also drop from sleeve_weights), then call `persist_paper_book(repo, capital=capital,
  run_date=as_of, sleeve_weights=…, symbol_weights_by_sleeve=…, prices=prices, at=at)` and return its count.
  (In production, `prices` + `sleeves` + `sleeve_weights` are assembled by the asset in Task 7; the service
  stays pure-ish + injectable for tests.)
- [ ] **Step 4:** PASS. **Step 5:** `git commit -m "feat(paper-snap): PaperSnapshotService (skip-on-fail)"`

---

## Task 7: Dagster asset `paper_book_snapshot` wired into the daily schedule

**Files:** Modify `src/fxhnt/adapters/orchestration/assets.py` + `definitions.py`;
Test `tests/integration/test_paper_snapshot_asset.py`

- [ ] **Step 1: Failing test** — materialize/call the plain builder with a fake warehouse + fake sleeves and
  assert it persists positions (mirror the existing `build_combined_forward_nav` test pattern in the repo).
- [ ] **Step 2:** run → FAIL.
- [ ] **Step 3:** Add `build_paper_book_snapshot(...)` (plain fn, unit-testable) + `@asset
  paper_book_snapshot` that `deps=[combined_forward_nav]`. It assembles inputs for `PaperSnapshotService`:
  the live sleeves (from the registry/composition), `current_sleeve_weights(...)` from the book's edge
  series, and latest closes per referenced symbol via `WarehousePriceProvider`. Construct `PaperRepo` from
  the same cockpit DSN as `ForwardNavRepo`. Guard the whole asset body in try/except so a snapshot failure
  logs but does NOT fail the run (it must not break `combined_forward_nav`). Add the asset to the
  `Definitions` in `definitions.py` so the daily 23:30 schedule materializes it after the book.
- [ ] **Step 4:** PASS; also `uv run pytest tests/integration -q -k "forward or book or dagster or definition"`
  to confirm the nightly graph still loads + no regression.
- [ ] **Step 5:** `git commit -m "feat(paper-snap): paper_book_snapshot Dagster asset (daily, after book)"`

---

## Task 8: Full suite + finish + deploy

- [ ] **Step 1:** `uv run pytest -q` → all green.
- [ ] **Step 2:** Use **superpowers:finishing-a-development-branch** — merge `feat/paper-snapshot` → master,
  push to Gitea.
- [ ] **Step 3:** Deploy: `./scripts/argo-deploy-cockpit.sh` (or via the Gitea webhook auto-deploy from the
  push). Verify the workflow Succeeded.
- [ ] **Step 4:** Trigger one snapshot (materialize the `paper_book_snapshot` asset once via Dagster, or wait
  for 23:30) and confirm `https://dashboard.fxhnt.ai/paper` (or the cockpit `/paper` route) shows real
  positions + live MTM. Report what renders.

---

## Self-review notes
- Per-sleeve `current_weights` MUST reuse the sleeve's existing domain fn (no re-derivation) — fidelity.
- A failing/no-data sleeve returns `{}` / is caught + skipped; never breaks the nightly book.
- Read-only/additive: `combined_forward_nav` + gate record unchanged.
- `do NOT git add -A` — `*.dbn` data on disk must not be staged (gitignored; verify).

## Acceptance criteria (from spec)
- each live sleeve has `current_weights(as_of)` reusing domain logic, unit-tested. ✅ T2,T3,T4
- allocator exposes `current_sleeve_weights`, unit-tested. ✅ T5
- `PaperSnapshotService.snapshot()` persists multi-sleeve + skips failures. ✅ T6
- `paper_book_snapshot` asset wired into daily schedule, deps combined_forward_nav, wiring-tested. ✅ T7
- post-deploy `/paper` shows real positions/trades/PnL. ✅ T8
- nightly book + gate record unchanged. ✅ (additive; asset try/except)
