# Bybit 4-edge Levered Shadow Track (Phase 1) 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:** Record an observe-only levered shadow of the `bybit_4edge` book (static differentiated leverage, book-gross ≤ 1.5x) alongside the honest naive track, so we learn live whether the levered book reconciles OOS — no capital, no adaptive re-tune, no killswitch.

**Architecture:** Reuse every existing part. The levered book is the SAME per-sleeve funding-net returns combined with a static per-sleeve leverage shape and a hard gross cap, instead of equal weight. One new pure combine function, a leverage-shape parameter threaded through the existing `BybitFourEdgeStrategy` / backtest-summary / forward-nav asset, a registry row, and one persist call. The nightly `ForwardTracker` + `cockpit_forward` + reconciliation gate all work unchanged.

**Tech Stack:** Python 3.13, SQLAlchemy (cockpit repos), Dagster assets, pytest, in-memory sqlite `TimescaleFeatureStore` for tests.

## Global Constraints

- Observe-only: NO capital deployment, NO change to the naive `bybit_4edge` track.
- Book-gross cap: `MAX_BOOK_GROSS = 1.5` (hard). Leverage shape (relative): `crypto_tstrend 1.0, unlock 1.5, xsfunding 5.0, positioning 2.0` — carry-weighted, scaled down to hit the 1.5x gross cap.
- The levered gate MUST get its own `backtest_summary` row (else the recon gate silently degrades — see `reference_fxhnt_forward_gate_two_store_sync`).
- The heavy full-history compute lives ONLY in the `bybit-persist-sleeve-ret` own-memory Job, never the OOM-prone dagster daemon.
- Never commit secrets; commit messages end `Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>`.

---

### Task 1: `levered_eqwt_daily_returns` — the gross-capped levered combine

**Files:**
- Modify: `src/fxhnt/application/bybit_forward_track.py` (add function + two module constants near `naive_eqwt_daily_returns`)
- Test: `tests/integration/test_bybit_forward_track.py`

**Interfaces:**
- Produces: `levered_eqwt_daily_returns(returns_by_sleeve: dict[str, dict[int, float]], sleeves: list[str], shape: dict[str, float], *, max_gross: float = 1.5, cost_bps: float = 0.0) -> list[tuple[int, float]]`; constants `_LEVERED_SLEEVE_SHAPE: dict[str,float]`, `MAX_BOOK_GROSS: float`.

- [ ] **Step 1: Write the failing tests**

```python
# in tests/integration/test_bybit_forward_track.py
def test_levered_eqwt_uniform_shape_equals_naive() -> None:
    """With a uniform shape (all 1.0) and max_gross >= 1, the levered combine reduces EXACTLY to the naive
    equal-weight book — same weights, same gross 1.0."""
    from fxhnt.application.bybit_forward_track import (
        levered_eqwt_daily_returns, naive_eqwt_daily_returns)
    rbs = {"a": {1: 0.01, 2: -0.02, 3: 0.03}, "b": {1: 0.00, 2: 0.04, 3: -0.01}}
    naive = naive_eqwt_daily_returns(rbs, ["a", "b"], cost_bps=0.0)
    lev = levered_eqwt_daily_returns(rbs, ["a", "b"], {"a": 1.0, "b": 1.0}, max_gross=1.5, cost_bps=0.0)
    assert lev == naive

def test_levered_eqwt_carry_weighted_and_gross_capped() -> None:
    """A carry-heavy shape puts more weight on the high-shape sleeve, and the book gross is capped at
    max_gross (the summed absolute weight never exceeds it)."""
    from fxhnt.application.bybit_forward_track import levered_eqwt_daily_returns
    rbs = {"trend": {1: 0.02, 2: 0.02}, "carry": {1: 0.02, 2: 0.02}}
    lev = levered_eqwt_daily_returns(rbs, ["trend", "carry"], {"trend": 1.0, "carry": 5.0},
                                     max_gross=1.5, cost_bps=0.0)
    # both sleeves move +2%; shape 1:5 -> carry weight 5x trend; scaled so gross = 1.5.
    # weights: trend=1/2, carry=5/2 -> gross=3.0 -> scale 0.5 -> trend=0.25, carry=1.25 (gross 1.5)
    # day return = 0.25*0.02 + 1.25*0.02 = 0.03
    assert abs(lev[0][1] - 0.03) < 1e-12
```

- [ ] **Step 2: Run to verify they fail**

Run: `uv run pytest tests/integration/test_bybit_forward_track.py -k "levered_eqwt" -q`
Expected: FAIL with `ImportError: cannot import name 'levered_eqwt_daily_returns'`.

- [ ] **Step 3: Implement the function + constants**

Add near the top of `bybit_forward_track.py` (after `_EPOCH`) the constants, and after `naive_eqwt_daily_returns` the function:

```python
# Phase-1 static leverage shape (relative per-sleeve leverage), scaled down to the book-gross cap. Carry is
# the low-vol workhorse; trend stays ~1x (its liquidation vector). See the 2026-07-03 leverage spec.
_LEVERED_SLEEVE_SHAPE: dict[str, float] = {
    "crypto_tstrend": 1.0, "unlock": 1.5, "xsfunding": 5.0, "positioning": 2.0}
MAX_BOOK_GROSS = 1.5


def levered_eqwt_daily_returns(returns_by_sleeve: dict[str, dict[int, float]], sleeves: list[str],
                              shape: dict[str, float], *, max_gross: float = 1.5,
                              cost_bps: float = 0.0) -> list[tuple[int, float]]:
    """The book's per-day return series `[(epoch_day, ret), ...]` under a STATIC per-sleeve leverage `shape`
    (relative leverage), scaled so the summed absolute weight (book gross) never exceeds `max_gross`. Mirrors
    `naive_eqwt_daily_returns` exactly (static weight over the present set, absent-today sleeve idles at 0,
    same turnover charge) — it IS the naive book when `shape` is uniform and `max_gross >= 1`."""
    present = [s for s in sleeves if returns_by_sleeve.get(s)]
    if not present:
        return []
    n = len(present)
    w = {s: shape.get(s, 1.0) / n for s in present}
    gross = sum(abs(v) for v in w.values())
    if gross > max_gross and gross > 0:
        scale = max_gross / gross
        w = {s: v * scale for s, v in w.items()}
    all_days = sorted({d for s in present for d in returns_by_sleeve[s]})
    daily: list[tuple[int, float]] = []
    prev_w = {s: 0.0 for s in present}
    for d in all_days:
        today = [s for s in present if d in returns_by_sleeve[s]]
        if not today:
            continue
        g = sum(w[s] * returns_by_sleeve[s][d] for s in today)
        if cost_bps:
            cur_w = {s: (w[s] if d in returns_by_sleeve[s] else 0.0) for s in present}
            turnover = sum(abs(cur_w[s] - prev_w[s]) for s in present)
            g -= turnover * cost_bps / 1e4
            prev_w = cur_w
        daily.append((d, g))
    return daily
```

- [ ] **Step 4: Run to verify they pass**

Run: `uv run pytest tests/integration/test_bybit_forward_track.py -k "levered_eqwt" -q`
Expected: PASS (2 passed).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/bybit_forward_track.py tests/integration/test_bybit_forward_track.py
git commit -m "feat(bybit-levered): gross-capped levered eq-wt combine (Phase 1)"
```

---

### Task 2: Thread `leverage_shape` through `BybitFourEdgeStrategy`

**Files:**
- Modify: `src/fxhnt/application/bybit_forward_track.py` (`BybitFourEdgeStrategy.__init__` + `.advance`)
- Test: `tests/integration/test_bybit_forward_track.py`

**Interfaces:**
- Consumes: `levered_eqwt_daily_returns`, `_LEVERED_SLEEVE_SHAPE`, `MAX_BOOK_GROSS` (Task 1).
- Produces: `BybitFourEdgeStrategy(store, *, universe=None, sleeves=None, cost_bps=5.5, unlock_events=None, stablecoin_spot_panel=None, leverage_shape: dict[str,float] | None = None, max_gross: float = MAX_BOOK_GROSS)`. `leverage_shape=None` → naive (unchanged); set → levered.

- [ ] **Step 1: Write the failing test**

```python
def test_bybit_four_edge_strategy_levered_differs_from_naive() -> None:
    """With a leverage_shape the strategy books the levered book (carry-weighted, gross-capped); without it,
    the naive book — same days, different returns."""
    from fxhnt.application.bybit_forward_track import BybitFourEdgeStrategy, _LEVERED_SLEEVE_SHAPE
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    _seed_momentum(store, n_symbols=8, days=260)
    _seed_carry(store, days=260)
    naive, _ = BybitFourEdgeStrategy(store, cost_bps=0.0).advance(None, {})
    lev, _ = BybitFourEdgeStrategy(store, cost_bps=0.0, leverage_shape=_LEVERED_SLEEVE_SHAPE).advance(None, {})
    store.close()
    assert [d for d, _ in lev] == [d for d, _ in naive]          # same day spine
    assert [r for _, r in lev] != [r for _, r in naive]          # different (levered) returns
```

- [ ] **Step 2: Run to verify it fails**

Run: `uv run pytest tests/integration/test_bybit_forward_track.py -k "strategy_levered_differs" -q`
Expected: FAIL with `TypeError: __init__() got an unexpected keyword argument 'leverage_shape'`.

- [ ] **Step 3: Implement — add the params + branch**

In `BybitFourEdgeStrategy.__init__`, add after `self._stablecoin_spot_panel = ...` (keep existing lines):

```python
        self._leverage_shape = leverage_shape
        self._max_gross = max_gross
```

and extend the signature with `leverage_shape: dict[str, float] | None = None, max_gross: float = MAX_BOOK_GROSS`.

In `.advance`, replace the `daily = naive_eqwt_daily_returns(...)` line with:

```python
        if self._leverage_shape is None:
            daily = naive_eqwt_daily_returns(returns_by_sleeve, active, cost_bps=self._cost_bps)
        else:
            daily = levered_eqwt_daily_returns(returns_by_sleeve, active, self._leverage_shape,
                                               max_gross=self._max_gross, cost_bps=self._cost_bps)
```

- [ ] **Step 4: Run to verify it passes**

Run: `uv run pytest tests/integration/test_bybit_forward_track.py -k "strategy_levered_differs or backtest_summary or naive_eqwt" -q`
Expected: PASS (naive path unchanged; levered path differs).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/bybit_forward_track.py tests/integration/test_bybit_forward_track.py
git commit -m "feat(bybit-levered): thread leverage_shape through BybitFourEdgeStrategy"
```

---

### Task 3: Parametrize `bybit_4edge_backtest_summary` for the levered strategy_id

**Files:**
- Modify: `src/fxhnt/application/bybit_forward_track.py` (`bybit_4edge_backtest_summary`)
- Test: `tests/integration/test_bybit_forward_track.py`

**Interfaces:**
- Consumes: Task 2's `leverage_shape`.
- Produces: `bybit_4edge_backtest_summary(store, *, universe=None, sleeves=None, cost_bps=5.5, unlock_events=None, stablecoin_spot_panel=None, leverage_shape=None, strategy_id="bybit_4edge")` — same return type `BacktestSummary`, `strategy_id` overridable.

- [ ] **Step 1: Write the failing test**

```python
def test_levered_backtest_summary_uses_levered_strategy_id_and_differs() -> None:
    """The levered backtest reference carries strategy_id 'bybit_4edge_levered' and a DIFFERENT cagr from the
    naive one (leverage scales the book), so the levered recon gate reconciles against the levered backtest."""
    from fxhnt.application.bybit_forward_track import bybit_4edge_backtest_summary, _LEVERED_SLEEVE_SHAPE
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    _seed_momentum(store, n_symbols=8, days=260)
    _seed_carry(store, days=260)
    naive = bybit_4edge_backtest_summary(store, cost_bps=0.0)
    lev = bybit_4edge_backtest_summary(store, cost_bps=0.0, leverage_shape=_LEVERED_SLEEVE_SHAPE,
                                       strategy_id="bybit_4edge_levered")
    store.close()
    assert lev.strategy_id == "bybit_4edge_levered"
    assert lev.cagr != naive.cagr
```

- [ ] **Step 2: Run to verify it fails**

Run: `uv run pytest tests/integration/test_bybit_forward_track.py -k "levered_backtest_summary_uses" -q`
Expected: FAIL with `TypeError: ... unexpected keyword argument 'leverage_shape'`.

- [ ] **Step 3: Implement — add params + thread them**

Extend `bybit_4edge_backtest_summary`'s signature with `leverage_shape: dict[str, float] | None = None, strategy_id: str = "bybit_4edge"`. Pass `leverage_shape=leverage_shape` into the `BybitFourEdgeStrategy(...)` construction inside it, and replace both literal `strategy_id="bybit_4edge"` occurrences (the empty-rows early return AND the final return) with `strategy_id=strategy_id`.

- [ ] **Step 4: Run to verify it passes**

Run: `uv run pytest tests/integration/test_bybit_forward_track.py -k "backtest_summary" -q`
Expected: PASS (naive summary unchanged; levered summary distinct).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/bybit_forward_track.py tests/integration/test_bybit_forward_track.py
git commit -m "feat(bybit-levered): parametrize backtest_summary for the levered strategy_id"
```

---

### Task 4: Registry row `bybit_4edge_levered`

**Files:**
- Modify: `src/fxhnt/registry.py` (add entry after `bybit_4edge`)
- Test: `tests/integration/test_bybit_forward_track.py`

**Interfaces:**
- Produces: `STRATEGY_REGISTRY["bybit_4edge_levered"]` with `state_file="bybit_4edge_levered_state"`, reconciliation gate `min_forward_days=21`.

- [ ] **Step 1: Write the failing test**

```python
def test_registry_has_bybit_4edge_levered_shadow() -> None:
    from fxhnt.registry import STRATEGY_REGISTRY
    e = STRATEGY_REGISTRY["bybit_4edge_levered"]
    assert e["state_file"] == "bybit_4edge_levered_state"
    assert e["venue"] == "bybit-perp"
    assert e["gate_spec"] == {"gate_type": "reconciliation", "min_forward_days": 21}
```

- [ ] **Step 2: Run to verify it fails**

Run: `uv run pytest tests/integration/test_bybit_forward_track.py -k "registry_has_bybit_4edge_levered" -q`
Expected: FAIL with `KeyError: 'bybit_4edge_levered'`.

- [ ] **Step 3: Implement — add the registry entry**

Immediately after the `"bybit_4edge": { ... },` block in `registry.py`, add:

```python
    # OBSERVE-ONLY shadow of bybit_4edge under STATIC differentiated leverage (carry-weighted, book-gross
    # <= 1.5x). Records what the levered book WOULD do so we can validate it OOS before any capital rides it
    # (Phase 1 of the 2026-07-03 leverage spec). Same reconciliation gate as the naive book.
    "bybit_4edge_levered": {
        "display_name": "Bybit 4-edge book (levered shadow, <=1.5x)", "sleeve": "crypto-bybit",
        "venue": "bybit-perp", "state_file": "bybit_4edge_levered_state",
        "tier": "research", "execution": "paper",
        "gate_spec": {"gate_type": "reconciliation", "min_forward_days": 21},
    },
```

- [ ] **Step 4: Run to verify it passes**

Run: `uv run pytest tests/integration/test_bybit_forward_track.py -k "registry_has_bybit_4edge_levered" -q`
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/registry.py tests/integration/test_bybit_forward_track.py
git commit -m "feat(bybit-levered): register bybit_4edge_levered shadow track"
```

---

### Task 5: Nightly asset `bybit_4edge_levered_nav` + wire into the job

**Files:**
- Modify: `src/fxhnt/adapters/orchestration/assets.py` (add `build_bybit_4edge_levered_forward_nav` next to `build_bybit_4edge_forward_nav`; add `bybit_4edge_levered_nav` asset next to `bybit_4edge_nav`; add it to `combined_book_forward_job` and `cockpit_forward`'s `deps`)
- Test: `tests/integration/test_bybit_forward_track.py`

**Interfaces:**
- Consumes: Task 2's `leverage_shape` param, `_LEVERED_SLEEVE_SHAPE`.
- Produces: `build_bybit_4edge_levered_forward_nav(store, state_path, *, universe=None, cost_bps=5.5, unlock_events=None, stablecoin_spot_panel=None) -> ForwardStatus`; Dagster asset `bybit_4edge_levered_nav`.

- [ ] **Step 1: Write the failing test (the plain builder, no Dagster)**

```python
def test_build_bybit_4edge_levered_forward_nav_books_the_levered_book(tmp_path) -> None:
    """The levered forward builder freezes T0 and books the levered (not naive) book to its OWN state file."""
    from fxhnt.adapters.orchestration.assets import build_bybit_4edge_levered_forward_nav
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    _seed_momentum(store, n_symbols=8, days=130)
    _seed_carry(store, days=130)
    state = str(tmp_path / "bybit_4edge_levered_state.json")
    st = build_bybit_4edge_levered_forward_nav(store, state, cost_bps=0.0)
    store.close()
    assert st.inception is not None and (tmp_path / "bybit_4edge_levered_state.json").exists()
```

- [ ] **Step 2: Run to verify it fails**

Run: `uv run pytest tests/integration/test_bybit_forward_track.py -k "build_bybit_4edge_levered" -q`
Expected: FAIL with `ImportError: cannot import name 'build_bybit_4edge_levered_forward_nav'`.

- [ ] **Step 3: Implement the plain builder**

Directly after `build_bybit_4edge_forward_nav` in `assets.py`, add:

```python
def build_bybit_4edge_levered_forward_nav(store, state_path, *, universe=None, cost_bps=5.5,
                                          unlock_events=None, stablecoin_spot_panel=None):  # type: ignore[no-untyped-def]
    """Observe-only LEVERED shadow of the 4-edge book: the SAME sleeves under the static `_LEVERED_SLEEVE_SHAPE`
    (carry-weighted, book-gross <= MAX_BOOK_GROSS), stepped into its OWN state file. No capital; records what
    the levered book would do so it can be validated OOS before deployment."""
    from fxhnt.application.bybit_forward_track import (
        MAX_BOOK_GROSS, _LEVERED_SLEEVE_SHAPE, BybitFourEdgeStrategy)
    from fxhnt.application.forward_tracker import ForwardTracker

    strategy = BybitFourEdgeStrategy(
        store, universe=universe, cost_bps=cost_bps, unlock_events=unlock_events,
        stablecoin_spot_panel=stablecoin_spot_panel,
        leverage_shape=_LEVERED_SLEEVE_SHAPE, max_gross=MAX_BOOK_GROSS)
    return ForwardTracker(strategy, state_path).step()
```

- [ ] **Step 4: Run to verify the builder test passes**

Run: `uv run pytest tests/integration/test_bybit_forward_track.py -k "build_bybit_4edge_levered" -q`
Expected: PASS.

- [ ] **Step 5: Add the Dagster asset + wire it in**

After the `bybit_4edge_nav` asset, add an asset that mirrors it exactly but calls the levered builder and writes `bybit_4edge_levered_state.json`:

```python
@asset(deps=[refresh_bybit_warehouse])
def bybit_4edge_levered_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """OBSERVE-ONLY levered shadow of bybit_4edge (static leverage, book-gross <= 1.5x). Steps its own
    ForwardTracker; forward-nav-ONLY (light, OOM-proof), same as bybit_4edge_nav."""
    import urllib.error

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
    from fxhnt.config import get_settings

    s = get_settings()
    store = TimescaleFeatureStore(s.operational_dsn, table="bybit_features")
    try:
        universe = liquid_universe(store) or None
        unlock_events = load_unlock_events(operational_unlock_repo(s), f"{_data_dir()}/unlock_calendar.json")
        st = build_bybit_4edge_levered_forward_nav(
            store, f"{_data_dir()}/bybit_4edge_levered_state.json",
            universe=universe, cost_bps=s.cost_bps_per_turnover, unlock_events=unlock_events)
        context.log.info(
            f"bybit_4edge_levered_nav: T0={st.inception}, {st.forward_days}d through {st.last_date} "
            f"(+{st.forward_return_pct:.2f}%, Sharpe {st.forward_sharpe:.2f})")
        return {"inception": st.inception, "last_date": st.last_date, "forward_days": st.forward_days,
                "forward_return_pct": st.forward_return_pct, "forward_sharpe": st.forward_sharpe}
    except (urllib.error.URLError, TimeoutError, ConnectionError, OSError) as e:
        context.log.warning(f"bybit_4edge_levered_nav step failed (state unchanged): {e}")
        return {"inception": None, "last_date": None, "forward_days": 0}
    finally:
        store.close()
```

Then add `bybit_4edge_levered_nav` to BOTH the `combined_book_forward_job` asset selection and the `cockpit_forward` `@asset(deps=[...])` list (the line beginning `@asset(deps=[combined_forward_nav, ...`), alongside the existing `bybit_4edge_nav`.

- [ ] **Step 6: Run the whole forward-track + orchestration test surface**

Run: `uv run pytest tests/integration/test_bybit_forward_track.py tests/integration/test_bybit_book_eval.py -q`
Expected: PASS (all).

- [ ] **Step 7: Commit**

```bash
git add src/fxhnt/adapters/orchestration/assets.py tests/integration/test_bybit_forward_track.py
git commit -m "feat(bybit-levered): nightly bybit_4edge_levered_nav shadow asset + job wiring"
```

---

### Task 6: Persist the levered backtest reference in the persist Job

**Files:**
- Modify: `src/fxhnt/cli.py` (`bybit_persist_sleeve_ret`, right after the existing naive `bybit_4edge_backtest_summary` upsert)
- Test: `tests/integration/test_bybit_forward_track.py`

**Interfaces:**
- Consumes: Task 3's `bybit_4edge_backtest_summary(..., leverage_shape=, strategy_id=)`.
- Produces: a `backtest_summary` row `strategy_id='bybit_4edge_levered'` so the levered recon gate reconciles (does not silently degrade).

- [ ] **Step 1: Write the failing test (the provider gets a levered reference)**

```python
def test_levered_backtest_ref_activates_the_levered_gate() -> None:
    """Persisting the levered summary gives the cockpit ref provider a real reference for
    'bybit_4edge_levered' — the levered gate's band is active, not the degraded no-reference timer."""
    from fxhnt.application.bybit_forward_track import bybit_4edge_backtest_summary, _LEVERED_SLEEVE_SHAPE
    from fxhnt.application.forward_ingest import CockpitBacktestRefProvider
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    _seed_momentum(store, n_symbols=8, days=260)
    _seed_carry(store, days=260)
    sm = bybit_4edge_backtest_summary(store, cost_bps=0.0, leverage_shape=_LEVERED_SLEEVE_SHAPE,
                                      strategy_id="bybit_4edge_levered")
    store.close()
    repo = ForwardNavRepo("sqlite://")
    repo.migrate()
    repo.upsert_backtest_summary(sm, at=dt.datetime(2026, 7, 3))
    ref = CockpitBacktestRefProvider(repo).reference("bybit_4edge_levered", 21)
    assert ref is not None and ref.expected_window_return != 0.0
```

- [ ] **Step 2: Run to verify it fails**

Run: `uv run pytest tests/integration/test_bybit_forward_track.py -k "levered_backtest_ref_activates" -q`
Expected: PASS already at the unit level (it exercises the pure function + repo directly). This test guards the MECHANISM; the CLI wiring in Step 3 is what makes it happen in prod. Run it now to confirm green, then add the CLI wiring so the prod Job actually writes the row.

- [ ] **Step 3: Implement — persist the levered summary in the CLI**

In `bybit_persist_sleeve_ret`, immediately after the existing `fwd_repo.upsert_backtest_summary(bt, ...)` line, add:

```python
        from fxhnt.application.bybit_forward_track import _LEVERED_SLEEVE_SHAPE
        bt_lev = bybit_4edge_backtest_summary(
            store, universe=universe, cost_bps=cost_bps, unlock_events=unlock_events,
            leverage_shape=_LEVERED_SLEEVE_SHAPE, strategy_id="bybit_4edge_levered")
        fwd_repo.upsert_backtest_summary(bt_lev, at=dt.datetime.now(dt.UTC).replace(tzinfo=None))
        log.info("bybit-persist-sleeve-ret: bybit_4edge_levered backtest ref cagr=%.1f%% -> backtest_summary",
                 bt_lev.cagr * 100)
```

(The `bybit_4edge_backtest_summary` import already exists a few lines above from the naive persist.)

- [ ] **Step 4: Run the full affected surface**

Run: `uv run pytest tests/integration/test_bybit_forward_track.py -q && uv run pytest tests/ -k "forward_ingest or reconciliation_gate" -q`
Expected: PASS (all).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/cli.py tests/integration/test_bybit_forward_track.py
git commit -m "feat(bybit-levered): persist bybit_4edge_levered backtest reference in the persist Job"
```

---

## Post-implementation (not tasks — deploy/verify, done together after review)

- Full suite: `uv run pytest tests/ -q` (expect all green).
- Deploy: commit → push master → build → dagster rollout (nightly runs `bybit_4edge_levered_nav`).
- Activate the reference now: run `bybit-persist-sleeve-ret` locally against prod (writes both naive + levered `backtest_summary` rows).
- Trigger `bybit_4edge_levered_nav` + `cockpit_forward` once to freeze T0 and populate the shadow row; verify `forward_summary` shows `bybit_4edge_levered` building, and `CockpitBacktestRefProvider(...).reference("bybit_4edge_levered", 21)` is non-None.
- Then: observe. The learning phase — compare the levered shadow vs the naive track live before any Phase 2 work.

## Self-Review

- **Spec coverage:** Phase 1 = Component 3 with a static leverage vector (spec's phasing) — Tasks 1–2 (static levered combine + strategy), Task 3 (levered backtest ref), Task 4 (registry/gate), Task 5 (nightly shadow track), Task 6 (reference persisted so the gate doesn't silently degrade). Components 1–2 + killswitch are explicitly Phase 2, not here. Observe-only / no-capital honored (no execution code). ✓
- **Placeholder scan:** none — every step has real code/commands.
- **Type consistency:** `leverage_shape: dict[str,float] | None` and `strategy_id: str` are used identically across Tasks 2, 3, 5, 6; `_LEVERED_SLEEVE_SHAPE` / `MAX_BOOK_GROSS` defined in Task 1 and imported everywhere after; `build_bybit_4edge_levered_forward_nav` signature matches the naive builder. ✓
