# Equity Leg Deployable Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: superpowers:subagent-driven-development. Checkbox (`- [ ]`) steps.

**Goal:** Make the IBKR US-ETF `multistrat` book honest-sized + capacity-bounded (real ADV from ingested ETF volume, real trading costs), so it deploys as the growth vehicle above the crypto book's ~$350k break.

**Architecture:** An equity analog of the crypto honest-reference. Ingest real Yahoo ETF volume (PIT); a real cost model (sqrt-impact on dollar-ADV + per-ETF spread + IBKR commission) net of `multistrat`'s turnover; a cost-netted `book_series` sizing series + ADV capacity curve; an `equity_allocation_inputs` provider merged into `record_allocation`.

**Tech Stack:** Python 3.12, SQLAlchemy 2.0, typer, pytest. Spec: `docs/superpowers/specs/2026-07-16-equity-leg-deployable-design.md`.

## Global Constraints

- `FUND_INSTRUMENTS = ["SPY","IEF","GLD","PDBC","DBMF"]` (from `multistrat`).
- **Per-ETF half-spread constants (bps), documented — the one estimated input** (no ETF bid/ask feed): `{"SPY":0.3,"IEF":1.0,"GLD":1.0,"PDBC":4.0,"DBMF":8.0}`. Default for an unlisted symbol: 5.0 bps.
- **IBKR commission**: `commission_per_share = 0.0035` USD; `commission_bps = 100 * commission_per_share / price` (tiny for high-priced ETFs — dominant cost is spread).
- **Dollar-ADV**: `adv_usd[sym][day] = pit_close[sym][day] * volume[sym][day]`, trailing **63-day MEDIAN** (outlier-robust; a volume spike must not inflate capacity). A day with missing volume is excluded (no fabricated ADV).
- **PIT discipline**: volume snapshot is first-seen-frozen (a re-fetch must NOT overwrite a stored (symbol,date) — mirror `snapshot_yahoo_closes`).
- **Byte-identity**: `record_allocation` with `store=None`, or with no equity sid in the deploy set, must be byte-identical to today. The equity provider only adds series/ceilings for `ibkr-equity` deploy sids.
- **Cost invariant**: costs must REDUCE the equity Sharpe (net ≤ gross); the turnover charged is the ACTUAL traded weight change (two-way), from `multistrat.target_weights`, not the raw signal.
- Reuse the crypto templates (`honest_sleeve_returns` shape, `capacity_ceiling_from_curve`, load-once/reprice). Do NOT reintroduce a flat 15bp cost.
- Commit trailer: `Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>`.

---

### Task 1: Ingest ETF volume (fetch + PIT store + snapshot + backfill CLI)

**Files:**
- Modify: `src/fxhnt/adapters/data/yahoo_daily.py` (`.volumes()`)
- Modify: `src/fxhnt/adapters/persistence/cockpit_models.py` (`YahooPitVolumeRow`)
- Modify: `src/fxhnt/adapters/persistence/forward_nav.py` (`snapshot_yahoo_volumes`, `yahoo_pit_volumes`)
- Modify: `src/fxhnt/adapters/orchestration/assets.py` (nightly snapshot of volumes)
- Modify: `src/fxhnt/cli.py` (`backfill-etf-volume`)
- Test: `tests/unit/test_yahoo_volume_pit.py`

**Interfaces:**
- Produces: `YahooDailyClient.volumes(symbol) -> dict[str,float]` (date→raw share volume); `ForwardNavRepo.snapshot_yahoo_volumes(symbol, series, at)` / `.yahoo_pit_volumes(symbol) -> dict[str,float]`.

- [ ] **Step 1: Write failing tests** — a fake Yahoo response dict → `volumes()` extracts `quote[0]["volume"]` mapped to dates (skipping None); a repo roundtrip: `snapshot_yahoo_volumes` then `yahoo_pit_volumes` returns the series; first-seen-frozen (a second snapshot with a changed value for an existing date does NOT overwrite).

```python
# tests/unit/test_yahoo_volume_pit.py
import datetime as dt
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo

def test_snapshot_and_read_volumes_first_seen_frozen():
    r = ForwardNavRepo("sqlite://"); r.migrate()
    at = dt.datetime(2026, 7, 16)
    r.snapshot_yahoo_volumes("SPY", {"2026-07-01": 1000.0, "2026-07-02": 2000.0}, at)
    r.snapshot_yahoo_volumes("SPY", {"2026-07-01": 9999.0, "2026-07-03": 3000.0}, at)  # 07-01 must NOT change
    assert r.yahoo_pit_volumes("SPY") == {"2026-07-01": 1000.0, "2026-07-02": 2000.0, "2026-07-03": 3000.0}

def test_volumes_extracts_quote_volume(monkeypatch):
    from fxhnt.adapters.data.yahoo_daily import YahooDailyClient
    c = YahooDailyClient()
    fake = {"chart": {"result": [{"timestamp": [1719792000, 1719878400],
            "indicators": {"quote": [{"volume": [111.0, None]}], "adjclose": [{"adjclose": [1.0, 2.0]}]}}]}}
    monkeypatch.setattr(c, "_get", lambda url: fake)
    out = c.volumes("SPY")
    assert list(out.values()) == [111.0]   # None volume day skipped
```

- [ ] **Step 2: Run — expect FAIL**

Run: `cd /home/jgrusewski/Work/fxhnt && .venv/bin/python3 -m pytest tests/unit/test_yahoo_volume_pit.py -q`
Expected: FAIL.

- [ ] **Step 3: Implement.**
  - `yahoo_daily.py` — `volumes(symbol)` mirrors `adj_closes` but reads `ind["quote"][0]["volume"]`:
    ```python
    def volumes(self, symbol: str) -> dict[str, float]:
        res = self._get(_BASE.format(sym=symbol, rng=self._rng))["chart"]["result"][0]
        ts = res["timestamp"]
        vol = res["indicators"]["quote"][0].get("volume") or []
        out: dict[str, float] = {}
        for t, v in zip(ts, vol, strict=False):
            if v is not None:
                out[dt.datetime.fromtimestamp(t, dt.timezone.utc).strftime("%Y-%m-%d")] = float(v)
        return out
    ```
  - `cockpit_models.py` — `YahooPitVolumeRow` mirroring `YahooPitCloseRow` (`__tablename__="yahoo_pit_volume"`, PK (symbol,date), `volume: Mapped[float]`, `first_seen_at`).
  - `forward_nav.py` — `snapshot_yahoo_volumes` / `yahoo_pit_volumes` mirroring the close methods exactly (first-seen-frozen).
  - `assets.py` — in the block that calls `repo.snapshot_yahoo_closes(sym, live.adj_closes(sym), at)` (~line 421), add `repo.snapshot_yahoo_volumes(sym, live.volumes(sym), at)` in the same loop (wrap the volume fetch in try/except so a volume-fetch failure never breaks the close snapshot — WARN + continue).
  - `cli.py` — `@app.command("backfill-etf-volume")`: for each `FUND_INSTRUMENTS` sym, `YahooDailyClient(rng="25y").volumes(sym)` → `repo.snapshot_yahoo_volumes(...)`; echo counts. (Import `FUND_INSTRUMENTS` from `fxhnt.domain.strategies.multistrat`.)

- [ ] **Step 4: Run — expect PASS**

Run: `cd /home/jgrusewski/Work/fxhnt && .venv/bin/python3 -m pytest tests/unit/test_yahoo_volume_pit.py -q`
Expected: PASS.

- [ ] **Step 5: Commit** (`git add` the 5 files + test; message `feat(equity): ingest ETF volume (PIT) + backfill-etf-volume CLI`).

---

### Task 2: `equity_honest.py` — real cost model + capacity curve

**Files:**
- Create: `src/fxhnt/application/equity_honest.py`
- Test: `tests/unit/test_equity_honest.py`

**Interfaces:**
- Consumes: `IMPACT_COEFF_BPS` (honest_reference), `sharpe_of`/`per_period_sharpe`, `capacity_ceiling_from_curve` (allocation_engine).
- Produces:
  - `ETF_HALF_SPREAD_BPS: dict[str,float]` (the documented constants) + `_spread_bps(sym) -> float`.
  - `dollar_adv(pit_closes, pit_volumes, window=63) -> dict[str, dict[str,float]]` (per-sym trailing-median dollar volume).
  - `equity_honest_returns(book_ret_by_day, weights_by_day, adv_usd, *, aum, participation_cap=0.03, impact_k=1.0, commission_per_share=0.0035) -> dict[str,float]`
  - `equity_capacity_curve(book_ret_by_day, weights_by_day, adv_usd, *, aums, ...) -> dict[float, dict[str,float]]`

- [ ] **Step 1: Write failing tests.**
  - `dollar_adv`: given closes + volumes over N days, the ADV is the trailing 63d (or shorter early) median of `close*volume`; a missing-volume day is excluded.
  - `equity_honest_returns`: (a) with ZERO turnover (constant weights) → net == gross book_ret (no cost). (b) with turnover + tiny AUM (adv >> notional) → cost ≈ spread-only (impact ~0), net < gross by `Σ traded_w * spread/1e4`. (c) net Sharpe ≤ gross Sharpe. (d) a position exceeding `participation_cap * adv` is scaled down (capacity), reducing that day's gross contribution.
  - `equity_capacity_curve`: monotonic-ish — higher AUM → lower or equal net Sharpe (impact grows); returns one series per aum.

```python
# tests/unit/test_equity_honest.py (sketch — fill real numbers)
from fxhnt.application.equity_honest import dollar_adv, equity_honest_returns, equity_capacity_curve, ETF_HALF_SPREAD_BPS

def test_zero_turnover_no_cost():
    days = ["2026-01-01","2026-01-02","2026-01-03"]
    book = {d: 0.001 for d in days}
    w = {d: {"SPY": 0.5} for d in days}                 # constant weights -> no turnover after day 1
    adv = {"SPY": {d: 1e12 for d in days}}              # infinite liquidity -> no impact
    net = equity_honest_returns(book, w, adv, aum=100_000.0)
    # day 1 enters the position (turnover from 0->0.5) -> small spread cost; days 2-3 no turnover -> net==book
    assert abs(net["2026-01-02"] - book["2026-01-02"]) < 1e-12
    assert abs(net["2026-01-03"] - book["2026-01-03"]) < 1e-12

def test_costs_reduce_sharpe():
    ...  # net sharpe <= gross sharpe over a turnover-y series
```

- [ ] **Step 2: Run — expect FAIL**

Run: `cd /home/jgrusewski/Work/fxhnt && .venv/bin/python3 -m pytest tests/unit/test_equity_honest.py -q`
Expected: FAIL.

- [ ] **Step 3: Implement `equity_honest.py`.** Mirror `honest_sleeve_returns`'s structure (turnover-basis cost, participation cap), but with equity inputs. Reference implementation:

```python
"""Equity honest-reference: the IBKR US-ETF book's cost-netted return series + capacity, the analog of the
crypto honest_reference. Real costs: sqrt market-impact on REAL dollar-ADV (ingested Yahoo volume) + a
documented per-ETF spread + IBKR commission, charged on the book's ACTUAL traded weight (two-way turnover).
Pure; no I/O (callers pass preloaded closes/volumes/weights)."""
from __future__ import annotations

import math
import statistics

from fxhnt.application.honest_reference import IMPACT_COEFF_BPS
from fxhnt.domain.gauntlet import per_period_sharpe

ETF_HALF_SPREAD_BPS: dict[str, float] = {"SPY": 0.3, "IEF": 1.0, "GLD": 1.0, "PDBC": 4.0, "DBMF": 8.0}
_DEFAULT_SPREAD_BPS = 5.0


def _spread_bps(sym: str) -> float:
    return ETF_HALF_SPREAD_BPS.get(sym, _DEFAULT_SPREAD_BPS)


def dollar_adv(pit_closes: dict[str, dict[str, float]], pit_volumes: dict[str, dict[str, float]],
               *, window: int = 63) -> dict[str, dict[str, float]]:
    """Per-symbol trailing-`window` MEDIAN dollar volume (close*volume). Missing-volume days excluded."""
    out: dict[str, dict[str, float]] = {}
    for sym, vols in pit_volumes.items():
        closes = pit_closes.get(sym, {})
        dv = {d: closes[d] * vols[d] for d in sorted(vols) if d in closes and vols[d] is not None}
        dates = sorted(dv)
        cur: dict[str, float] = {}
        for i, d in enumerate(dates):
            lo = max(0, i - window + 1)
            cur[d] = statistics.median(dv[x] for x in dates[lo:i + 1])
        out[sym] = cur
    return out


def equity_honest_returns(book_ret_by_day: dict[str, float], weights_by_day: dict[str, dict[str, float]],
                          adv_usd: dict[str, dict[str, float]], *, aum: float,
                          participation_cap: float = 0.03, impact_k: float = 1.0,
                          commission_per_share: float = 0.0035,
                          price_by_day: dict[str, dict[str, float]] | None = None) -> dict[str, float]:
    """Cost-netted daily book return. Cost = Σ_sym traded_w * (max(impact, spread) + commission)/1e4, on the
    two-way traded weight; positions exceeding participation_cap*ADV are scaled down (capacity)."""
    out: dict[str, float] = {}
    prev: dict[str, float] = {}
    for day in sorted(book_ret_by_day):
        row = weights_by_day.get(day, {})
        cur: dict[str, float] = {}
        scale_hit = 0.0
        for sym, w in row.items():
            adv = adv_usd.get(sym, {}).get(day, 0.0)
            notional = abs(w) * aum
            if adv <= 0.0:
                cur[sym] = 0.0            # no ADV -> zero capacity (honest, not infinite)
                scale_hit += abs(w)       # this weight can't be held -> its return is forgone
            elif notional > participation_cap * adv:
                cur[sym] = w * (participation_cap * adv) / notional
                scale_hit += abs(w) - abs(cur[sym])
            else:
                cur[sym] = w
        cost = 0.0
        for sym in set(cur) | set(prev):
            traded_w = abs(cur.get(sym, 0.0) - prev.get(sym, 0.0))
            if traded_w <= 0.0:
                continue
            adv = adv_usd.get(sym, {}).get(day, 0.0)
            traded_notional = traded_w * aum
            impact = impact_k * IMPACT_COEFF_BPS * math.sqrt(traded_notional / adv) if adv > 0.0 and traded_notional > 0.0 else 0.0
            price = (price_by_day or {}).get(sym, {}).get(day, 0.0)
            commission_bps = (100.0 * commission_per_share / price) if price > 0.0 else 0.0
            slip_bps = max(impact, _spread_bps(sym)) + commission_bps
            cost += traded_w * (slip_bps / 1e4)
        # capacity-forgone return: reduce the day's gross by the scaled-out fraction of the book's own return
        out[day] = book_ret_by_day[day] * (1.0 - scale_hit) - cost if scale_hit < 1.0 else -cost
        prev = cur
    return out


def equity_capacity_curve(book_ret_by_day, weights_by_day, adv_usd, *, aums,
                          participation_cap: float = 0.03, impact_k: float = 1.0,
                          commission_per_share: float = 0.0035, price_by_day=None):
    """{aum: equity_honest_returns(...at aum)} — load-once inputs, reprice per AUM."""
    return {aum: equity_honest_returns(book_ret_by_day, weights_by_day, adv_usd, aum=aum,
                                       participation_cap=participation_cap, impact_k=impact_k,
                                       commission_per_share=commission_per_share, price_by_day=price_by_day)
            for aum in aums}
```

**Implementer note:** the `scale_hit` capacity-forgone term is an approximation (scales the whole book's return by the un-held fraction). Confirm it is monotone (higher AUM → more scale_hit → lower net) and document it. If a cleaner per-symbol attribution is available from `book_series`'s internals, prefer it — but do NOT block on it; the book-level approximation is acceptable for v1 and must be stated.

- [ ] **Step 4: Run — expect PASS.** Run the test file; fix until green.

- [ ] **Step 5: Commit** (`feat(equity): real ETF cost model + ADV capacity curve (equity_honest.py)`).

---

### Task 3: `equity_allocation_inputs` provider + `record_allocation` merge

**Files:**
- Create: `src/fxhnt/application/equity_allocation_inputs.py`
- Modify: `src/fxhnt/application/allocation_ingest.py` (merge equity series/ceilings)
- Test: `tests/unit/test_equity_allocation_inputs.py`, `tests/integration/test_record_allocation_equity.py`

**Interfaces:**
- Consumes: `MultiStratStrategy` + `book_series`/`target_weights` (multistrat), `YahooPitClient` + `yahoo_pit_volumes` (repo), `equity_honest.*`, `capacity_ceiling_from_curve`.
- Produces: `equity_allocation_inputs(repo, deploy_sids, *, target_aum, aums, participation_cap=0.03, impact_k=1.0, min_sharpe=1.0) -> tuple[dict[str,dict[str,float]], dict[str,float|None]]`.

- [ ] **Step 1: Confirm the multistrat shapes.** Read `multistrat.book_series` and `multistrat.target_weights` signatures/returns and `MultiStratStrategy.advance` — determine the exact type of the daily book-return series and the weights vector (keys = ISO date? epoch? sym→weight?). The provider must feed `equity_honest_returns` `book_ret_by_day: {isodate: ret}` and `weights_by_day: {isodate: {sym: weight}}`. Adapt keys as needed (ISO strings, matching the nav record for `deploy_book_returns` alignment).

- [ ] **Step 2: Write failing tests.** A fake repo returning seeded `yahoo_pit_closes`/`yahoo_pit_volumes` for FUND_INSTRUMENTS over enough days; `equity_allocation_inputs(repo, {"multistrat"}, target_aum=100_000, aums=[...])` returns a non-empty ISO-keyed series for `multistrat` (len ≥ 60) + a finite ceiling. And in `record_allocation`: with an equity deploy sid present + store injected, `multistrat` gets a series (is eligible, not dropped); with `store=None` OR no equity sid, the result is byte-identical to before (add a regression assertion).

- [ ] **Step 3: Run — expect FAIL.**

- [ ] **Step 4: Implement `equity_allocation_inputs.py`:**
  - For each equity deploy sid (registry `venue == "ibkr-equity"`; today only `multistrat`):
    - `closes = {sym: repo.yahoo_pit_closes(sym) for sym in FUND_INSTRUMENTS}`; `volumes = {sym: repo.yahoo_pit_volumes(sym) for sym in FUND_INSTRUMENTS}`.
    - `book_ret = book_series(closes, ...)` → `{isodate: ret}`; `weights = target_weights(closes, ...)` → `{isodate: {sym: w}}` (adapt from Step 1).
    - `adv = dollar_adv(closes, volumes)`.
    - base series = `equity_honest_returns(book_ret, weights, adv, aum=target_aum, price_by_day=closes)`; store ISO-keyed under the sid.
    - ceiling = `capacity_ceiling_from_curve({aum: sharpe_of(equity_honest_returns(...at aum)) for aum in aums}, min_sharpe)` via `equity_capacity_curve` (load-once).
  - Absent/short PIT data → no series (caller falls back), ceiling `None`.
- [ ] **Step 4b: Wire `record_allocation`** (`allocation_ingest.py`, in the `store is not None` branch after `honest_allocation_inputs`):

```python
        from fxhnt.application.equity_allocation_inputs import equity_allocation_inputs
        eq_sids = {sid for sid in funded_sids
                   if STRATEGY_REGISTRY.get(sid, {}).get("venue") == "ibkr-equity"}
        if eq_sids:
            eq_returns, eq_ceilings = equity_allocation_inputs(
                repo, eq_sids, target_aum=equity,
                aums=[100_000, 1_000_000, 10_000_000, 50_000_000],
                min_sharpe=getattr(pol, "capacity_min_sharpe", 1.0))
            honest_returns = {**honest_returns, **eq_returns}
            ceilings = {**ceilings, **eq_ceilings}
```

(`repo` is the `ForwardNavRepo` already constructed at the top of `record_allocation`.)

- [ ] **Step 5: Run — expect PASS** (unit + integration; existing allocation tests still green — byte-identity for no-equity/store=None).

Run: `cd /home/jgrusewski/Work/fxhnt && .venv/bin/python3 -m pytest tests/unit/test_equity_allocation_inputs.py tests/integration/test_record_allocation_equity.py tests/unit -k "allocation or record" -q`

- [ ] **Step 6: Commit** (`feat(equity): equity_allocation_inputs provider merged into record_allocation`).

---

### Task 4: full-suite green

- [ ] **Step 1:** `cd /home/jgrusewski/Work/fxhnt && .venv/bin/python3 -m pytest tests/unit -q` → all pass.
- [ ] **Step 2:** `.venv/bin/python3 -m pytest tests/integration -k "allocation or honest or equity or multistrat or yahoo" -q` → all pass.
- [ ] **Step 3:** commit any fixes (else skip).

## Post-implementation (controller-run, in-cluster)
Deploy (git-sync) + `fxhnt migrate-cockpit` (creates `yahoo_pit_volume`) + `fxhnt backfill-etf-volume`. Then the §5 gates: G1 multistrat sizes (cost-netted Sharpe < gross), G2 real ADV ceiling (finite, DBMF/PDBC-bound), G3 cost realism (SPY/IEF/GLD effective bps << 15), G4 combined marginal gate {crypto book, multistrat} — report the honest verdict + combined deployment.

## Self-Review
- Coverage: §4a→T1, §4b→T2, §4c→T2, §4d→T3, gates→post-impl.
- Byte-identity: equity merge only fires for ibkr-equity deploy sids with store injected; no-equity/store=None unchanged (T3 regression test).
- Reuse: IMPACT_COEFF_BPS, capacity_ceiling_from_curve, load-once/reprice, first-seen PIT.
- The estimated spread constants + the `scale_hit` capacity approximation are documented, not hidden.
