# Paper book: vol-target leverage + drawdown killswitch — Implementation Plan (Phase 2F-A3d)

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

**Goal:** Layer the live `book_allocator` risk overlay onto A3c's sleeve weights: vol-target + fractional-Kelly
leverage and a drawdown killswitch, driven off a kill-proof per-sleeve return signal, applied identically in
the backfill and the nightly snapshot.

**Architecture:** New pure fns in `application/paper_book.py` — `vol_target_leverage`, `DrawdownKillSwitch`,
`paper_book_returns`, and a single `paper_risk_overlay(returns, active) -> (weights, leverage, killed)` both
paths call. A `leverage` arg on `persist_paper_book` scales notional only. A new `paper_shadow_positions` unit
book (never flattened) makes `sleeve_daily_return` survive kill periods so re-entry + vol estimates stay
correct. Read-only/additive; default `leverage=1.0` leaves every existing test unchanged.

**Tech Stack:** Python 3.12, pytest, SQLAlchemy, Dagster. **Repo:** `~/Work/fxhnt`, branch
`feat/paper-vol-target`. **Spec:** `docs/superpowers/specs/2026-06-22-paper-vol-target-killswitch-design.md`.

**Verified context:**
- `book_allocator.combine_book` step 4: `target_vol_daily = target_vol/√ppy`; `lev = min(MAX_LEVERAGE,
  kelly·target_vol_daily/pv)` where `pv = std` of the trailing weighted-portfolio return; `_MAX_LEVERAGE=3.0`.
  `apply_drawdown_killswitch`: shadow_eq cumulative `∏(1+r)`, kill when `dd>kill_dd`, re-enter when
  `dd<reenter_dd`. Defaults: `target_vol=0.12, kelly_fraction=0.5, vol_lookback=60, periods_per_year=365,
  kill_dd=0.15, reenter_dd=0.07`.
- `domain.paper.target_positions(sleeve, sleeve_weight, capital, symbol_weights, prices, date)` → notional =
  `sleeve_weight·capital`; `Position(sleeve, symbol, qty, entry_price, entry_date)`. `sleeve_daily_return`
  (A3c) is leverage/weight-invariant.
- `persist_paper_book(repo, *, capital, run_date, sleeve_weights, symbol_weights_by_sleeve, prices, at,
  enabled=True)` → `PaperBookService(repo, capital).derive_and_persist(...)` (uses
  `target_positions(..., capital=self._capital ...)`) + writes `paper_nav`. `equal_weights` is now dead (A3c).
- A3c helpers present: `isv_adaptive_weights(returns_by_sleeve, active)`, `sleeve_daily_return(prev, closes)`;
  `PaperRepo.sleeve_returns(before=)`/`upsert_sleeve_ret`, `positions_before(run_date)`, `replace_positions`.
- Backfill/snapshot bodies (post-A3c) compute the clean return from `positions_before` — A3d switches that to
  `shadow_positions_before` and adds the overlay + shadow persistence.

---

## Task 1: `vol_target_leverage` pure fn

**Files:** Modify `src/fxhnt/application/paper_book.py`; Test `tests/unit/test_vol_target_leverage.py`

- [ ] **Step 1: Failing test**
```python
# tests/unit/test_vol_target_leverage.py
import math
from fxhnt.application.paper_book import vol_target_leverage

def _series(vals, start=1):
    return {start + i: v for i, v in enumerate(vals)}

def test_no_history_cold_start_is_unlevered():
    assert vol_target_leverage({"a": {}}, {"a": {}}) == 1.0          # weights, returns both empty -> 1.0

def test_lower_vol_gives_more_leverage():
    lo = vol_target_leverage({"a": 1.0}, {"a": _series([0.002, -0.002] * 40)})
    hi = vol_target_leverage({"a": 1.0}, {"a": _series([0.02, -0.02] * 40)})
    assert lo > hi

def test_capped_at_max_leverage():
    tiny = vol_target_leverage({"a": 1.0}, {"a": _series([1e-6, -1e-6] * 40)}, max_leverage=3.0)
    assert tiny == 3.0

def test_formula_matches_book_allocator():
    # single sleeve weight 1.0, known std -> lev = min(MAX, kelly*(target_vol/sqrt(ppy))/pv)
    import numpy as np
    r = _series([0.01, -0.01, 0.005, -0.005] * 20)
    w = {"a": 1.0}
    lev = vol_target_leverage(w, {"a": r}, target_vol=0.12, kelly_fraction=0.5,
                              periods_per_year=365, vol_lookback=60, max_leverage=3.0)
    vals = [r[d] for d in sorted(r)][-60:]
    pv = float(np.std(vals)); tvd = 0.12 / math.sqrt(365)
    assert math.isclose(lev, min(3.0, 0.5 * tvd / pv), rel_tol=1e-9)
```
- [ ] **Step 2:** Run → FAIL.
- [ ] **Step 3:** Implement in `paper_book.py` (after `isv_adaptive_weights`; add module constants near the
  A3c constants `_L_MIN…`):
```python
_TARGET_VOL = 0.12
_KELLY_FRACTION = 0.5
_MAX_LEVERAGE = 3.0
_VOL_LOOKBACK = 60
_PERIODS_PER_YEAR = 365
_KILL_DD = 0.15
_REENTER_DD = 0.07


def vol_target_leverage(weights: dict[str, float], returns_by_sleeve: dict[str, dict[int, float]], *,
                        target_vol: float = _TARGET_VOL, kelly_fraction: float = _KELLY_FRACTION,
                        vol_lookback: int = _VOL_LOOKBACK, periods_per_year: int = _PERIODS_PER_YEAR,
                        max_leverage: float = _MAX_LEVERAGE) -> float:
    """Fractional-Kelly vol-target leverage over the trailing weighted-portfolio return (mirrors
    book_allocator.combine_book step 4). lev = min(max_leverage, kelly·(target_vol/√ppy)/pv) where pv is the
    std of Σ wᵢ·rᵢ over the last `vol_lookback` of the union date axis. Cold-start (pv≤1e-12 or <2 obs) → 1.0
    (unlevered = A3c behavior; never flat from lack of data). Pure."""
    days = sorted({d for s in weights for d in returns_by_sleeve.get(s, {})})
    if len(days) < 2:
        return 1.0
    win = days[max(0, len(days) - vol_lookback):]
    port = [sum(weights.get(s, 0.0) * returns_by_sleeve.get(s, {}).get(d, 0.0) for s in weights) for d in win]
    pv = float(np.std(port)) if len(port) > 1 else 0.0
    if pv <= 1e-12:
        return 1.0
    return min(max_leverage, kelly_fraction * (target_vol / math.sqrt(periods_per_year)) / pv)
```
  (Add `import math` if not present — check the top of the file; numpy `np` already imported by A3c.)
- [ ] **Step 4:** Run → PASS.
- [ ] **Step 5:** Commit `git commit -m "feat(paper): vol_target_leverage (fractional-Kelly vol-target)"`

---

## Task 2: `DrawdownKillSwitch` class

**Files:** Modify `src/fxhnt/application/paper_book.py`; Test `tests/unit/test_drawdown_killswitch.py`

- [ ] **Step 1: Failing test**
```python
# tests/unit/test_drawdown_killswitch.py
from fxhnt.application.paper_book import DrawdownKillSwitch

def test_not_killed_before_threshold():
    ks = DrawdownKillSwitch(kill_dd=0.15, reenter_dd=0.07)
    ks.observe(-0.10)            # 10% dd < 15%
    assert ks.killed is False

def test_kills_past_threshold():
    ks = DrawdownKillSwitch(kill_dd=0.15, reenter_dd=0.07)
    ks.observe(-0.20)            # 20% dd > 15%
    assert ks.killed is True

def test_reenters_after_recovery_with_hysteresis():
    ks = DrawdownKillSwitch(kill_dd=0.15, reenter_dd=0.07)
    ks.observe(-0.20); assert ks.killed is True       # killed
    ks.observe(0.05);  assert ks.killed is True        # dd still ~16% > 7% -> stay killed (hysteresis)
    ks.observe(0.12);  assert ks.killed is False       # recovered inside 7% -> re-enter

def test_decision_uses_prior_state_only():
    # .killed read BEFORE observe(next) is the state applying to the next date
    ks = DrawdownKillSwitch(kill_dd=0.15, reenter_dd=0.07)
    assert ks.killed is False    # initial: not killed
    ks.observe(-0.20)
    assert ks.killed is True
```
- [ ] **Step 2:** Run → FAIL.
- [ ] **Step 3:** Implement in `paper_book.py` (mirrors `apply_drawdown_killswitch`):
```python
class DrawdownKillSwitch:
    """Drawdown kill-switch with hysteresis (mirrors book_allocator.apply_drawdown_killswitch). Tracks the
    UN-throttled shadow equity; `.killed` (read before `observe`-ing the next return) is the state to apply to
    the next date. kill when running dd > kill_dd; re-enter when dd < reenter_dd. Causal; pure (no I/O)."""

    def __init__(self, kill_dd: float = _KILL_DD, reenter_dd: float = _REENTER_DD) -> None:
        self.kill_dd = kill_dd
        self.reenter_dd = reenter_dd
        self.shadow_eq = 1.0
        self.peak = 1.0
        self.killed = False

    def observe(self, book_return: float) -> None:
        self.shadow_eq *= (1.0 + book_return)
        self.peak = max(self.peak, self.shadow_eq)
        dd = (self.peak - self.shadow_eq) / self.peak if self.peak > 0 else 0.0
        if not self.killed and dd > self.kill_dd:
            self.killed = True
        elif self.killed and dd < self.reenter_dd:
            self.killed = False
```
- [ ] **Step 4:** Run → PASS.
- [ ] **Step 5:** Commit `git commit -m "feat(paper): DrawdownKillSwitch (hysteresis, mirrors allocator)"`

---

## Task 3: `paper_book_returns` + `paper_risk_overlay` pure fns

**Files:** Modify `src/fxhnt/application/paper_book.py`; Test `tests/unit/test_paper_risk_overlay.py`

- [ ] **Step 1: Failing test**
```python
# tests/unit/test_paper_risk_overlay.py
from fxhnt.application.paper_book import paper_book_returns, paper_risk_overlay

def _series(vals, start=1):
    return {start + i: v for i, v in enumerate(vals)}

def test_book_returns_are_causal_and_keyed_by_day():
    # one sleeve, weight ~1; book return on day d uses weights/leverage held from before d
    r = _series([0.01, -0.01] * 30)
    out = paper_book_returns({"a": r})
    assert isinstance(out, dict) and len(out) >= 1
    # first day has no prior-held weights -> not in output (no book return yet)
    assert min(r) not in out

def test_overlay_returns_weights_leverage_killed():
    r = _series([0.01, -0.01] * 30)
    w, lev, killed = paper_risk_overlay({"a": r}, ["a"])
    assert set(w) == {"a"} and w["a"] == 1.0
    assert lev > 0.0
    assert killed in (True, False)

def test_overlay_kills_after_sustained_drawdown():
    # a long losing streak in the (single-sleeve) book -> killed True
    losing = _series([-0.03] * 40)
    _, _, killed = paper_risk_overlay({"a": losing}, ["a"])
    assert killed is True

def test_overlay_not_killed_on_calm_history():
    calm = _series([0.001, -0.001] * 40)
    _, _, killed = paper_risk_overlay({"a": calm}, ["a"])
    assert killed is False
```
- [ ] **Step 2:** Run → FAIL.
- [ ] **Step 3:** Implement in `paper_book.py` (after `vol_target_leverage` + `DrawdownKillSwitch`):
```python
def paper_book_returns(returns_by_sleeve: dict[str, dict[int, float]], *,
                       target_vol: float = _TARGET_VOL, kelly_fraction: float = _KELLY_FRACTION,
                       vol_lookback: int = _VOL_LOOKBACK, periods_per_year: int = _PERIODS_PER_YEAR,
                       max_leverage: float = _MAX_LEVERAGE) -> dict[int, float]:
    """UN-throttled paper book daily return series (the killswitch's regime signal). Ascending over the union
    date axis; weights/leverage are recomputed from returns STRICTLY BEFORE each date and held into it, so
    bret(d) = lev_held·Σ w_held·r_d. Causal — no look-ahead. The historical 'active' set is every sleeve with
    prior return history (a faithful proxy; identical in both call paths). Pure."""
    all_days = sorted({d for r in returns_by_sleeve.values() for d in r})
    out: dict[int, float] = {}
    w_held: dict[str, float] | None = None
    lev_held = 1.0
    for d in all_days:
        if w_held is not None:
            out[d] = lev_held * sum(w_held.get(s, 0.0) * returns_by_sleeve.get(s, {}).get(d, 0.0)
                                    for s in w_held)
        before_d = {s: {k: v for k, v in r.items() if k < d} for s, r in returns_by_sleeve.items()}
        active = [s for s, r in before_d.items() if r]
        w_held = isv_adaptive_weights(before_d, active)
        lev_held = vol_target_leverage(w_held, before_d, target_vol=target_vol,
                                       kelly_fraction=kelly_fraction, vol_lookback=vol_lookback,
                                       periods_per_year=periods_per_year, max_leverage=max_leverage)
    return out


def paper_risk_overlay(returns_by_sleeve: dict[str, dict[int, float]], active_sleeves: list[str], *,
                       target_vol: float = _TARGET_VOL, kelly_fraction: float = _KELLY_FRACTION,
                       vol_lookback: int = _VOL_LOOKBACK, periods_per_year: int = _PERIODS_PER_YEAR,
                       max_leverage: float = _MAX_LEVERAGE, kill_dd: float = _KILL_DD,
                       reenter_dd: float = _REENTER_DD) -> tuple[dict[str, float], float, bool]:
    """Single entry point for one date's risk overlay given its trailing (strictly-prior) returns. Returns
    (ISV weights over active, vol-target leverage, killed). `killed` replays a DrawdownKillSwitch over the
    un-throttled book-return history → the state applying to THIS date. Both backfill and snapshot call this
    on their trailing window so live /paper == /paper/replay. Pure."""
    weights = isv_adaptive_weights(returns_by_sleeve, active_sleeves)
    leverage = vol_target_leverage(weights, returns_by_sleeve, target_vol=target_vol,
                                   kelly_fraction=kelly_fraction, vol_lookback=vol_lookback,
                                   periods_per_year=periods_per_year, max_leverage=max_leverage)
    ks = DrawdownKillSwitch(kill_dd=kill_dd, reenter_dd=reenter_dd)
    bret = paper_book_returns(returns_by_sleeve, target_vol=target_vol, kelly_fraction=kelly_fraction,
                              vol_lookback=vol_lookback, periods_per_year=periods_per_year,
                              max_leverage=max_leverage)
    for d in sorted(bret):
        ks.observe(bret[d])
    return weights, leverage, ks.killed
```
- [ ] **Step 4:** Run → PASS.
- [ ] **Step 5:** Commit `git commit -m "feat(paper): paper_book_returns + paper_risk_overlay (causal risk overlay)"`

---

## Task 4: `paper_shadow_positions` table + repo methods

**Files:** Modify `src/fxhnt/adapters/persistence/cockpit_models.py`,
`src/fxhnt/adapters/persistence/paper_repo.py`; Test `tests/integration/test_paper_shadow_positions.py`

- [ ] **Step 1: Failing test**
```python
# tests/integration/test_paper_shadow_positions.py
import datetime as dt
from fxhnt.adapters.persistence.paper_repo import PaperRepo
from fxhnt.domain.paper import Position

def _repo():
    r = PaperRepo("sqlite://"); r.migrate(); return r

def test_replace_and_read_shadow_before():
    r = _repo(); at = dt.datetime(2026, 1, 2, tzinfo=dt.UTC)
    r.replace_shadow_positions("2026-01-01",
        [Position("ts", "BTCUSDT", 0.01, 100.0, "2026-01-01")], at=at)
    r.replace_shadow_positions("2026-01-02",
        [Position("ts", "BTCUSDT", 0.02, 110.0, "2026-01-02")], at=at)
    before2 = r.shadow_positions_before("2026-01-02")
    assert len(before2) == 1 and before2[0].entry_price == 100.0    # the prior date's shadow book
    assert r.shadow_positions_before("2026-01-01") == []            # nothing strictly before

def test_replace_is_idempotent_per_run_date():
    r = _repo(); at = dt.datetime(2026, 1, 2, tzinfo=dt.UTC)
    r.replace_shadow_positions("2026-01-01", [Position("ts", "BTCUSDT", 0.01, 100.0, "2026-01-01")], at=at)
    r.replace_shadow_positions("2026-01-01", [Position("ts", "ETHUSDT", 0.5, 50.0, "2026-01-01")], at=at)
    before = r.shadow_positions_before("2026-01-02")
    assert len(before) == 1 and before[0].symbol == "ETHUSDT"        # replaced, not appended
```
- [ ] **Step 2:** Run → FAIL.
- [ ] **Step 3a:** Add model to `cockpit_models.py` (after `PaperSleeveRetRow`):
```python
class PaperShadowPositionRow(CockpitBase):
    """Unit (sleeve_weight=1, capital=1) sleeve composition per run_date — the kill-PROOF return signal.
    Persisted every date and NEVER flattened by the killswitch, so per-sleeve returns survive kill periods
    (re-entry detection + ISV vol estimate stay correct). (run_date, sleeve, symbol) PK — idempotent."""
    __tablename__ = "paper_shadow_positions"
    run_date: Mapped[dt.date] = mapped_column(Date, primary_key=True)
    sleeve: Mapped[str] = mapped_column(String(32), primary_key=True)
    symbol: Mapped[str] = mapped_column(String(32), primary_key=True)
    qty: Mapped[float] = mapped_column(Float)
    entry_price: Mapped[float] = mapped_column(Float)
    entry_date: Mapped[str] = mapped_column(String(10))
    src_updated_at: Mapped[dt.datetime] = mapped_column(DateTime)
```
- [ ] **Step 3b:** In `paper_repo.py` add `PaperShadowPositionRow` to the cockpit_models import, and add (after
  `shadow`-free position methods, e.g. near `positions_before`):
```python
    def replace_shadow_positions(self, run_date: str, positions: list[Position], *, at: dt.datetime) -> None:
        """Idempotent per run_date: rewrite that date's UNIT shadow book (never throttled)."""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            s.execute(delete(PaperShadowPositionRow).where(PaperShadowPositionRow.run_date == rd))
            for p in positions:
                s.add(PaperShadowPositionRow(
                    run_date=rd, sleeve=p.sleeve, symbol=p.symbol, qty=p.qty,
                    entry_price=p.entry_price, entry_date=p.entry_date, src_updated_at=at))
            s.commit()

    def shadow_positions_before(self, run_date: str) -> list[Position]:
        """The UNIT shadow book as of the most recent run_date STRICTLY BEFORE `run_date`."""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            prev = s.scalar(select(PaperShadowPositionRow.run_date)
                            .where(PaperShadowPositionRow.run_date < rd)
                            .order_by(PaperShadowPositionRow.run_date.desc()).limit(1))
            if prev is None:
                return []
            rows = s.scalars(select(PaperShadowPositionRow)
                             .where(PaperShadowPositionRow.run_date == prev)
                             .order_by(PaperShadowPositionRow.sleeve, PaperShadowPositionRow.symbol))
            return [Position(r.sleeve, r.symbol, r.qty, r.entry_price, r.entry_date) for r in rows]
```
  (`delete`, `select`, `Session`, `Position` already imported in paper_repo.py — confirm.)
- [ ] **Step 4:** Run → PASS. **Step 5:** Regression `uv run pytest tests/integration/test_paper_repo.py -q`.
- [ ] **Step 6:** Commit `git commit -m "feat(paper): paper_shadow_positions unit book (kill-proof return signal)"`

---

## Task 5: `persist_paper_book(leverage=…)` + `unit_shadow_positions` helper

**Files:** Modify `src/fxhnt/application/paper_book.py`; Test extend `tests/integration/test_paper_hook.py`
(or `tests/unit/`), add `tests/unit/test_unit_shadow_positions.py`

- [ ] **Step 1: Failing tests**
```python
# tests/unit/test_unit_shadow_positions.py
from fxhnt.application.paper_book import unit_shadow_positions

def test_builds_unit_notional_composition():
    swb = {"ts": {"BTCUSDT": 1.0}, "sc": {"USDTUSD": -1.0}}
    prices = {"BTCUSDT": 100.0, "USDTUSD": 1.0}
    out = unit_shadow_positions(swb, prices, "2026-01-01")
    by = {(p.sleeve, p.symbol): p for p in out}
    assert by[("ts", "BTCUSDT")].qty == 1.0 / 100.0      # w*1*1 / px
    assert by[("sc", "USDTUSD")].qty == -1.0 / 1.0
    assert all(p.entry_date == "2026-01-01" for p in out)
```
```python
# add to tests/integration/test_paper_hook.py
def test_persist_paper_book_leverage_scales_notional():
    import datetime as dt
    from fxhnt.adapters.persistence.paper_repo import PaperRepo
    from fxhnt.application.paper_book import persist_paper_book
    repo = PaperRepo("sqlite://"); repo.migrate()
    at = dt.datetime(2026, 1, 1, tzinfo=dt.UTC)
    kw = dict(capital=100_000.0, run_date="2026-01-01",
              sleeve_weights={"ts": 1.0}, symbol_weights_by_sleeve={"ts": {"BTCUSDT": 1.0}},
              prices={"BTCUSDT": 100.0}, at=at)
    persist_paper_book(repo, **kw, leverage=2.0)
    pos = repo.positions_at("2026-01-01")
    assert len(pos) == 1 and pos[0].qty == 2.0 * 100_000.0 / 100.0    # notional doubled
    nav = repo.nav_history()[-1]
    assert nav.equity == 100_000.0                                    # equity base unscaled (no pnl yet)
```
- [ ] **Step 2:** Run → FAIL.
- [ ] **Step 3a:** Add `unit_shadow_positions` to `paper_book.py` (near `equal_weights`):
```python
def unit_shadow_positions(symbol_weights_by_sleeve: dict[str, dict[str, float]],
                          prices: dict[str, float], date: str) -> list[Position]:
    """The UNIT (sleeve_weight=1, capital=1) composition for every active sleeve — the kill-proof signal
    book. Reuses target_positions so qty/entry semantics match the throttled book. Pure."""
    out: list[Position] = []
    for sleeve, sw in symbol_weights_by_sleeve.items():
        out.extend(target_positions(sleeve=sleeve, sleeve_weight=1.0, capital=1.0,
                                    symbol_weights=sw, prices=prices, date=date))
    return out
```
- [ ] **Step 3b:** Add a `leverage` param to `persist_paper_book` and `PaperBookService.derive_and_persist`:
  - `persist_paper_book(..., enabled: bool = True, leverage: float = 1.0)` → pass `leverage` into
    `PaperBookService(repo, capital).derive_and_persist(..., leverage=leverage)`.
  - In `derive_and_persist(self, run_date, sleeve_weights, symbol_weights_by_sleeve, prices, at, leverage: float = 1.0)`:
    change the `target_positions(...)` call to use `capital=self._capital * leverage` (leave the
    `paper_nav` equity computation using `self._capital` unchanged — base stays unscaled):
```python
            target.extend(target_positions(
                sleeve=sleeve, sleeve_weight=sw, capital=self._capital * leverage,
                symbol_weights=symbol_weights, prices=prices, date=run_date))
```
- [ ] **Step 4:** Run → PASS. **Step 5:** Regression — `uv run pytest tests/integration/test_paper_hook.py
  tests/integration/test_paper_backfill.py tests/integration/test_paper_snapshot.py -q` (default `leverage=1.0`
  must leave all green).
- [ ] **Step 6:** Commit `git commit -m "feat(paper): persist_paper_book leverage (notional only) + unit_shadow_positions"`

---

## Task 6: backfill — risk overlay + shadow book + clean signal from shadow

**Files:** Modify `src/fxhnt/application/paper_backfill.py`; Test extend `tests/integration/test_paper_backfill.py`

- [ ] **Step 1: Failing test** — a kill period flattens the throttled book while the shadow book + signal
  continue, and a later recovery restores positions:
```python
def test_backfill_killswitch_flattens_then_reenters():
    import datetime as dt
    from fxhnt.adapters.persistence.paper_repo import PaperRepo
    from fxhnt.application.paper_backfill import backfill_paper_book

    class Sleeve:
        def __init__(self, sym): self.sym = sym
        def current_weights(self, d): return {self.sym: 1.0}

    repo = PaperRepo("sqlite://"); repo.migrate()
    dates = [f"2026-{(1+(i//28)):02d}-{(1+(i%28)):02d}" for i in range(112)]  # ~4 months
    closes = {}
    px = 100.0
    for i, d in enumerate(dates):
        # calm up, then a deep crash (drives dd>15%), then recovery
        if i < 40:   px *= 1.001
        elif i < 70: px *= 0.985        # ~36% drawdown over 30 days -> kill fires
        else:        px *= 1.01         # recovery -> eventual re-enter
        closes[d] = {"X": px}
    n = backfill_paper_book(repo, capital=100_000.0, sleeves={"s": Sleeve("X")},
                            dates=dates, closes_by_date=closes, at=dt.datetime(2026, 6, 1, tzinfo=dt.UTC))
    assert n == len(dates)
    # shadow signal is continuous (no gaps) across the whole window
    assert len(repo.sleeve_returns(before=dates[-1]).get("s", {})) > 80
    # at least one date during the crash has a FLAT throttled book (killed) ...
    flat = [d for d in dates[55:75] if repo.positions_at(d) == []]
    assert flat, "expected a killed/flat throttled book during the crash"
    # ... while the shadow book for those same dates is non-empty (signal survives)
    assert all(repo.shadow_positions_before(d) for d in flat)
```
- [ ] **Step 2:** Run → FAIL.
- [ ] **Step 3:** Edit `backfill_paper_book`:
  - Import: `from fxhnt.application.paper_book import (paper_risk_overlay, persist_paper_book,
    sleeve_daily_return, unit_shadow_positions)` (drop `isv_adaptive_weights`).
  - Replace the sizing + return-booking block with:
```python
        active = list(symbol_weights_by_sleeve)
        weights, lev_raw, killed = paper_risk_overlay(returns_by_sleeve, active)
        eff_lev = 0.0 if killed else lev_raw
        persist_paper_book(
            repo, capital=capital, run_date=date,
            sleeve_weights=weights, symbol_weights_by_sleeve=symbol_weights_by_sleeve,
            prices=closes_by_date.get(date, {}), at=at, leverage=eff_lev)
        # Persist the UNIT shadow book for this date (never flattened) so the per-sleeve return signal
        # survives kill periods. Then book the clean per-sleeve return from the PRIOR shadow book marked to
        # this date's close — feeds the next date's window. Causal.
        prices_d = closes_by_date.get(date, {})
        repo.replace_shadow_positions(date, unit_shadow_positions(symbol_weights_by_sleeve, prices_d, date), at=at)
        if prices_d:
            prior = repo.shadow_positions_before(date)
            by_sleeve: dict[str, list] = {}
            for p in prior:
                by_sleeve.setdefault(p.sleeve, []).append(p)
            for sname, plist in by_sleeve.items():
                r = sleeve_daily_return(plist, prices_d)
                if r is not None:
                    repo.upsert_sleeve_ret(date, sname, r, at=at)
        persisted += 1
    return persisted
```
  - Update the module/function docstring: ISV weights + vol-target leverage + killswitch; clean signal from
    the unit shadow book.
- [ ] **Step 4:** Run → PASS (`uv run pytest tests/integration/test_paper_backfill.py -q`). Update any
  pre-existing test that asserted the A3c return-source (positions_before) if it now differs; report it.
- [ ] **Step 5:** Commit `git commit -m "feat(paper): backfill applies risk overlay + unit shadow signal"`

---

## Task 7: snapshot — same overlay + shadow + clean signal

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

- [ ] **Step 1: Failing test** — seed a losing trailing signal so `paper_risk_overlay` reports `killed`, then
  assert the snapshot persists a FLAT throttled book (kill applied) while still persisting the unit shadow
  book + a new `paper_sleeve_ret` row:
```python
def test_snapshot_applies_killswitch_when_drawdown():
    import datetime as dt
    from fxhnt.adapters.persistence.paper_repo import PaperRepo
    from fxhnt.application.paper_snapshot import PaperSnapshotService
    repo = PaperRepo("sqlite://"); repo.migrate()
    at = dt.datetime(2026, 2, 1, tzinfo=dt.UTC)
    base = dt.date(2026, 1, 1).toordinal()
    for k in range(40):                          # sustained losses -> killed
        d = dt.date.fromordinal(base + k).isoformat()
        repo.upsert_sleeve_ret(d, "s", -0.03, at=at)
    class Sleeve:
        def current_weights(self, d): return {"X": 1.0}
    svc = PaperSnapshotService(repo, capital=100_000.0, sleeves={"s": Sleeve()},
                               sleeve_weights={}, prices={"X": 100.0})
    as_of = dt.date.fromordinal(base + 40).isoformat()
    n = svc.snapshot(as_of=as_of, at=at)
    assert n == 0                                # killed -> flat throttled book
    assert repo.positions_at(as_of) == []
    # shadow book + signal still recorded for as_of
    assert any(p.sleeve == "s" for p in repo.shadow_positions_before(dt.date.fromordinal(base + 41).isoformat()))
```
- [ ] **Step 2:** Run → FAIL.
- [ ] **Step 3:** Edit `PaperSnapshotService.snapshot`:
  - Import: `from fxhnt.application.paper_book import (paper_risk_overlay, persist_paper_book,
    sleeve_daily_return, unit_shadow_positions)` (drop `isv_adaptive_weights`).
  - Replace the sizing + return-booking block with:
```python
        returns_by_sleeve = self._repo.sleeve_returns(before=as_of)
        active = list(symbol_weights_by_sleeve)
        weights, lev_raw, killed = paper_risk_overlay(returns_by_sleeve, active)
        eff_lev = 0.0 if killed else lev_raw
        n = persist_paper_book(
            self._repo, capital=self._capital, run_date=as_of,
            sleeve_weights=weights, symbol_weights_by_sleeve=symbol_weights_by_sleeve,
            prices=self._prices, at=at, enabled=self._enabled, leverage=eff_lev)
        if self._enabled:
            self._repo.replace_shadow_positions(
                as_of, unit_shadow_positions(symbol_weights_by_sleeve, self._prices, as_of), at=at)
            if self._prices:
                prior = self._repo.shadow_positions_before(as_of)
                by_sleeve: dict[str, list] = {}
                for p in prior:
                    by_sleeve.setdefault(p.sleeve, []).append(p)
                for sname, plist in by_sleeve.items():
                    r = sleeve_daily_return(plist, self._prices)
                    if r is not None:
                        self._repo.upsert_sleeve_ret(as_of, sname, r, at=at)
        return n
```
  - Update the method docstring (overlay + shadow signal; live == replay).
- [ ] **Step 4:** Run → PASS (`uv run pytest tests/integration/test_paper_snapshot.py
  tests/integration/test_paper_snapshot_asset.py -q`). Update pre-existing tests if the return-source change
  shifts an assertion; report it.
- [ ] **Step 5:** Commit `git commit -m "feat(paper): snapshot applies risk overlay + unit shadow signal"`

---

## Task 8: Full suite + finish + deploy + re-backfill + verify

- [ ] **Step 1:** `cd ~/Work/fxhnt && uv run pytest -q` → all green. Fix any A3c-era assertion the
  return-source/leverage change shifts.
- [ ] **Step 2:** `git status --short | grep -i '\.dbn'` must print nothing.
- [ ] **Step 3:** **superpowers:finishing-a-development-branch** — merge `feat/paper-vol-target` → master,
  push (auto-deploys via Gitea→Argo). Confirm the cockpit workflow Succeeded + dashboard/dagster pods rolled.
- [ ] **Step 4:** Re-run backfill in-cluster:
  `kubectl exec deploy/dagster -c daemon -n foxhunt -- fxhnt paper-backfill --days 365`.
- [ ] **Step 5: VERIFY** — leverage + killswitch visible and drawdown capped:
```bash
kubectl exec deploy/dagster -c daemon -n foxhunt -- python -c "
from sqlalchemy import create_engine, text
from fxhnt.config import get_settings
e=create_engine(get_settings().operational_dsn)
with e.connect() as c:
    print('shadow rows:', list(c.execute(text('select count(*) from paper_shadow_positions'))))
    print('flat (killed) throttled dates:', list(c.execute(text('select count(*) from paper_nav n where not exists (select 1 from paper_positions p where p.run_date=n.run_date)'))))
    print('nav span + final:', list(c.execute(text('select round(min(equity)::numeric,0) lo, round(max(equity)::numeric,0) hi, count(*) n from paper_nav'))))
    print('max drawdown vs A3c -29%:', list(c.execute(text('with e as (select run_date, equity, max(equity) over (order by run_date) pk from paper_nav) select round(min((equity-pk)/pk)::numeric,3) maxdd from e'))))
"
```
  Confirm: `paper_shadow_positions` populated; some killed/flat dates exist during drawdowns; the
  `maxdd` is shallower than A3c's −0.29. Load `https://dashboard.fxhnt.ai/paper` + `/paper/replay` and report
  the levered, kill-aware curve.

---

## Self-review notes
- `paper_risk_overlay` is the ONE entry point both paths call → live `/paper` == `/paper/replay`.
- Clean signal from the UNIT shadow book (never flattened) → re-entry detection + ISV vol estimate survive
  kills. `sleeve_daily_return` is leverage/weight-invariant so the shadow value matches the throttled value on
  non-killed days.
- Causality: weights/leverage/kill all from strictly-prior returns; `bret(d)` uses prior-held weights.
- `leverage` scales notional only; equity base stays `capital`; default 1.0 keeps existing callers/tests green.
- Read-only/additive; no change to `book_allocator`, sleeve run()/advance(), combine_book, gate record, brokers.
- NEVER `git add -A`; `*.dbn` gitignored — verify nothing stray staged before each commit.

## Acceptance criteria (from spec)
- `vol_target_leverage` formula/cap/cold-start. ✅ T1
- `DrawdownKillSwitch` hysteresis + causal. ✅ T2
- `paper_book_returns`/`paper_risk_overlay` causal `(weights, leverage, killed)`. ✅ T3
- `paper_shadow_positions` table + repo; idempotent. ✅ T4
- `persist_paper_book(leverage=)` scales notional not base; default 1.0 green. ✅ T5
- backfill + snapshot apply overlay + unit shadow signal; kill flattens actual book, shadow continues. ✅ T6, T7
- re-backfill → leverage + capped DD; shadow populated; /paper + /paper/replay render. ✅ T8
- full suite green; read-only/additive; no `*.dbn`. ✅ T8
