# IBKR Consolidation (OPRA-backed XSP-VRP + UCITS multistrat routing) 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:** Build the `vrp` equity-VRP sleeve on real OPRA XSP option data (recomputable, gated, CVaR-sized) and add PRIIPs-legal UCITS execution routing to the multistrat book — both execution legs coded but shipped suspended.

**Architecture:** Component A adds an options data path to the Databento adapter (OPRA.PILLAR), freezes XSP put bars into a PIT warehouse table, and a `VrpStrategy` recomputes a defined-risk put-credit-spread ladder off the frozen bars (put-call-parity forward → IV-inversion 20Δ strike → real-mark P&L), wired to the forward engine + reconciliation gate + merged CVaR sizing + cockpit. Component B leaves the multistrat modeled track untouched and adds a US→UCITS instrument map applied by a new `--venue ucits` execute path. Neither exec leg can be paper-tested on the current account, so both ship suspended.

**Tech Stack:** Python 3.12, pydantic v2, SQLAlchemy 2.0 (CockpitBase declarative), Dagster assets, `databento` 0.79 (OPRA.PILLAR), `ib_async` (IBKR options `mleg`), pytest.

## Global Constraints

- **Branch:** `feat/ibkr-consolidation-xsp-vrp-ucits`. Never commit to `master`.
- **Determinism spine:** the strategy recomputes ONLY off frozen PIT rows in `xsp_option_bars` — never call OPRA live inside `VrpStrategy` or `run_track`. OPRA is hit only by the freeze step (asset/backfill Job).
- **Cost guard:** every OPRA fetch estimates cost via `metadata.get_cost` and refuses above the cap (`DatabentoCostExceeded`) BEFORE downloading. Default cap `0.50`; the one-time backfill Job runs under a raised cap, chunked by month.
- **Defined-risk ONLY:** every spread is a put-credit-spread with bounded max loss `(width − credit) × 100`; `margin = width_points × 100 = $500` at `width_points=5.0`. Never a naked put.
- **VRP registry values (verbatim):** id `vrp`, `venue` `"ibkr-xsp"`, `sleeve` `"equity-vol"`, `tier` `"research"`, `execution` `"paper"`, `state_file` `"vrp_state"`, `record_mode` `"recompute"`, `gate_spec {"gate_type": "reconciliation", "min_forward_days": 21}`, `promote_on_pass` **False**.
- **Both exec CronJobs ship `suspend: true`.** `execute-multistrat` default `--venue us` leaves today's armed US-ETF IBKR-paper track untouched.
- **Freeze policy (fixed → deterministic):** the freeze step stores, per trading day, all XSP puts with DTE in `[20, 45]` and moneyness `strike/forward ∈ [0.80, 1.02]`. `VrpStrategy` selects and marks entirely from this frozen slice.
- **No heavy DDL/backfill in cockpit startup** — the 2013→ backfill is a separate Job, not app startup.
- **Real money errs false-WAIT:** `promote_on_pass=False`; verify the cockpit LOCALLY (read NAV/gate in a browser), never off synthetic/prod. Every review Minor is a latent bug — fix in-loop.

---

## File Structure

| File | Responsibility |
|---|---|
| `src/fxhnt/application/vrp_pricing.py` (new) | Pure BS/IV/delta + put-call-parity forward helpers (no I/O) |
| `src/fxhnt/adapters/data/databento.py` (modify) | `AssetClass.OPTION` branch + `fetch_option_chain` / `fetch_option_bars` (OPRA.PILLAR) |
| `src/fxhnt/domain/models.py` (modify) | `AssetClass.OPTION` enum member + `OptionContract` dataclass |
| `src/fxhnt/adapters/persistence/cockpit_models.py` (modify) | `XspOptionBarRow` table |
| `src/fxhnt/adapters/persistence/xsp_option_bars.py` (new) | `XspOptionBarsRepo` (`upsert_bars` / `bars_for` / `chain_asof`) |
| `src/fxhnt/application/vrp_book.py` (new) | `VrpStrategy` (defined-risk XSP put-spread ladder, recompute off frozen bars) |
| `src/fxhnt/registry.py` (modify) | `"vrp"` entry |
| `src/fxhnt/adapters/orchestration/migration_builders.py` (modify) | `builders["vrp"]` anchor seed |
| `src/fxhnt/adapters/orchestration/assets.py` (modify) | `vrp_nav` asset (freeze slice → run_track + persist_backtest_ref) |
| `src/fxhnt/adapters/orchestration/definitions.py` (modify) | `vrp_nav` into the job + cockpit_forward deps |
| `src/fxhnt/adapters/broker/ibkr.py` (modify) | `place_combo` XSP put-credit-spread `mleg` (suspended path) |
| `src/fxhnt/application/vrp_exec_record.py` (new) | `plan_and_record_vrp` (executed-vs-modeled twin) |
| `src/fxhnt/cli.py` (modify) | `execute-vrp` command; `--venue us\|ucits` on `execute-multistrat` |
| `src/fxhnt/config.py` (modify) | `UcitsSettings` (US→UCITS map) |
| `src/fxhnt/application/ucits_map.py` (new) | `apply_ucits_map(weights, prices, mapping)` + unmapped refusal |
| `infra/k8s/jobs/fxhnt-vrp-rebalancer.yaml` (new) | suspended weekly VRP CronJob + NetworkPolicy |
| `infra/k8s/jobs/fxhnt-ucits-rebalancer.yaml` (new) | suspended monthly UCITS CronJob |
| `infra/k8s/jobs/fxhnt-opra-backfill.yaml` (new) | one-time OPRA 2013→ backfill Job (raised cap, chunked) |

---

## Task 1: VRP pricing helpers (pure, no I/O)

**Files:**
- Create: `src/fxhnt/application/vrp_pricing.py`
- Test: `tests/unit/test_vrp_pricing.py`

**Interfaces:**
- Produces: `bs_put(F: float, K: float, sigma: float, tau: float, disc: float = 1.0) -> float`; `implied_vol_put(price: float, F: float, K: float, tau: float, disc: float = 1.0) -> float | None`; `put_delta(F: float, K: float, sigma: float, tau: float) -> float` (returns `|Δ|` in `[0,1]`); `parity_forward(call_atm: float, put_atm: float, k_atm: float, disc: float = 1.0) -> float`.

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

```python
# tests/unit/test_vrp_pricing.py
import math
import pytest
from fxhnt.application.vrp_pricing import bs_put, implied_vol_put, put_delta, parity_forward


def test_bs_put_atm_positive_and_bounded():
    # ATM put (F=K) has value between 0 and K; ~0.4*F*sigma*sqrt(tau) for small tau
    p = bs_put(F=100.0, K=100.0, sigma=0.20, tau=30 / 365)
    assert 0.0 < p < 100.0
    assert p == pytest.approx(0.4 * 100.0 * 0.20 * math.sqrt(30 / 365), rel=0.05)


def test_bs_put_monotone_in_strike():
    lo = bs_put(F=100.0, K=90.0, sigma=0.2, tau=0.1)
    hi = bs_put(F=100.0, K=110.0, sigma=0.2, tau=0.1)
    assert hi > lo  # higher strike put is worth more


def test_implied_vol_round_trip():
    price = bs_put(F=100.0, K=95.0, sigma=0.27, tau=45 / 365)
    iv = implied_vol_put(price=price, F=100.0, K=95.0, tau=45 / 365)
    assert iv == pytest.approx(0.27, abs=1e-4)


def test_implied_vol_returns_none_on_arbitrage():
    # a price below intrinsic (K-F, discounted) has no arb-free IV
    assert implied_vol_put(price=0.0001, F=100.0, K=130.0, tau=0.1) is None


def test_put_delta_otm_near_target():
    # ~20-delta OTM put sits below spot; delta magnitude in (0,0.5)
    d = put_delta(F=100.0, K=94.0, sigma=0.20, tau=30 / 365)
    assert 0.0 < d < 0.5


def test_parity_forward_recovers_forward():
    F = 100.0
    tau = 0.1
    sigma = 0.2
    # build synthetic ATM call/put at K=100 consistent with forward F via parity: C - P = F - K
    p = bs_put(F=F, K=100.0, sigma=sigma, tau=tau)
    c = p + (F - 100.0)
    assert parity_forward(call_atm=c, put_atm=p, k_atm=100.0) == pytest.approx(F, abs=1e-6)
```

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

Run: `.venv/bin/python -m pytest tests/unit/test_vrp_pricing.py -q`
Expected: FAIL (`ModuleNotFoundError: fxhnt.application.vrp_pricing`).

- [ ] **Step 3: Implement the helpers**

```python
# src/fxhnt/application/vrp_pricing.py
"""Pure Black–Scholes helpers for the defined-risk XSP VRP sleeve — priced under the FORWARD measure so
the underlying comes from put-call parity on the option chain (no external index feed; OPRA is options-only).
No I/O: every input is a real OPRA mark or a frozen bar. r is folded into `disc = e^{-rT}` and the forward F."""
from __future__ import annotations

import math

_SQRT2 = math.sqrt(2.0)


def _norm_cdf(x: float) -> float:
    return 0.5 * (1.0 + math.erf(x / _SQRT2))


def bs_put(F: float, K: float, sigma: float, tau: float, disc: float = 1.0) -> float:
    """European put under the forward measure: disc*(K*N(-d2) - F*N(-d1)). disc = e^{-rT} (1.0 for r≈0)."""
    if tau <= 0.0 or sigma <= 0.0 or F <= 0.0 or K <= 0.0:
        return max(0.0, disc * (K - F))                      # intrinsic (degenerate day)
    vol = sigma * math.sqrt(tau)
    d1 = (math.log(F / K) + 0.5 * sigma * sigma * tau) / vol
    d2 = d1 - vol
    return disc * (K * _norm_cdf(-d2) - F * _norm_cdf(-d1))


def put_delta(F: float, K: float, sigma: float, tau: float) -> float:
    """|Δ| of the put under the forward measure = N(-d1) in [0,1]. Used for ~20-delta strike selection."""
    if tau <= 0.0 or sigma <= 0.0 or F <= 0.0 or K <= 0.0:
        return 1.0 if K > F else 0.0
    vol = sigma * math.sqrt(tau)
    d1 = (math.log(F / K) + 0.5 * sigma * sigma * tau) / vol
    return _norm_cdf(-d1)


def implied_vol_put(price: float, F: float, K: float, tau: float, disc: float = 1.0) -> float | None:
    """Invert bs_put for sigma via bisection. Returns None when `price` is below intrinsic / above the bound
    (no arb-free vol) — the caller skips that contract for the day."""
    if tau <= 0.0 or price <= 0.0:
        return None
    intrinsic = max(0.0, disc * (K - F))
    upper_bound = disc * K                                    # put <= disc*K (sigma -> inf)
    if price < intrinsic - 1e-9 or price > upper_bound + 1e-9:
        return None
    lo, hi = 1e-4, 5.0
    if bs_put(F, K, hi, tau, disc) < price:                  # unreachable even at 500% vol
        return None
    for _ in range(80):
        mid = 0.5 * (lo + hi)
        if bs_put(F, K, mid, tau, disc) < price:
            lo = mid
        else:
            hi = mid
    return 0.5 * (lo + hi)


def parity_forward(call_atm: float, put_atm: float, k_atm: float, disc: float = 1.0) -> float:
    """Implied forward from put-call parity: C - P = disc*(F - K) => F = K + (C - P)/disc."""
    return k_atm + (call_atm - put_atm) / disc
```

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

Run: `.venv/bin/python -m pytest tests/unit/test_vrp_pricing.py -q`
Expected: PASS (6 passed).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/vrp_pricing.py tests/unit/test_vrp_pricing.py
git commit -m "feat(vrp): pure BS/IV/delta + put-call-parity forward helpers"
```

---

## Task 2: Databento OPRA options path

**Files:**
- Modify: `src/fxhnt/domain/models.py` (add `AssetClass.OPTION`, `OptionContract`)
- Modify: `src/fxhnt/adapters/data/databento.py:47-76` (`_resolve` OPTION branch + two fetch methods)
- Test: `tests/unit/test_databento_options.py`

**Interfaces:**
- Consumes: `DatabentoCostExceeded`, `metadata.get_cost` guard (existing).
- Produces: `OptionContract(osi_symbol: str, expiry: dt.date, strike: float, right: str)` (frozen dataclass, `right` ∈ `{"P","C"}`); `DatabentoDataProvider.fetch_option_chain(underlying_parent: str, as_of: str) -> list[OptionContract]`; `DatabentoDataProvider.fetch_option_bars(osi_symbols: list[str], start: str, end: str) -> dict[str, dict[str, float]]` (osi → {iso_date: close}).

- [ ] **Step 1: Write the failing tests** (parse recorded-shape DataFrames, not a live pull)

```python
# tests/unit/test_databento_options.py
import datetime as dt
import pandas as pd
import pytest
from fxhnt.adapters.data.databento import DatabentoDataProvider, _chain_from_definition_df, _bars_from_ohlcv_df


def test_chain_from_definition_df_parses_puts_and_calls():
    df = pd.DataFrame(
        {
            "raw_symbol": ["XSP   240816P00470000", "XSP   240816C00470000"],
            "expiration": [pd.Timestamp("2024-08-16", tz="UTC"), pd.Timestamp("2024-08-16", tz="UTC")],
            "strike_price": [470.0, 470.0],
            "instrument_class": ["P", "C"],
        }
    )
    chain = _chain_from_definition_df(df)
    assert len(chain) == 2
    put = next(c for c in chain if c.right == "P")
    assert put.strike == 470.0 and put.expiry == dt.date(2024, 8, 16)
    assert put.osi_symbol == "XSP   240816P00470000"


def test_bars_from_ohlcv_df_groups_by_symbol():
    idx = pd.DatetimeIndex([pd.Timestamp("2024-07-15", tz="UTC"), pd.Timestamp("2024-07-16", tz="UTC")])
    df = pd.DataFrame(
        {"symbol": ["XSP   240816P00470000", "XSP   240816P00470000"], "close": [1.20, 1.35]}, index=idx
    )
    bars = _bars_from_ohlcv_df(df)
    assert bars["XSP   240816P00470000"] == {"2024-07-15": 1.20, "2024-07-16": 1.35}
```

- [ ] **Step 2: Run to verify failure**

Run: `.venv/bin/python -m pytest tests/unit/test_databento_options.py -q`
Expected: FAIL (`ImportError: cannot import name '_chain_from_definition_df'`).

- [ ] **Step 3: Add the enum + dataclass**

In `src/fxhnt/domain/models.py`, add to `AssetClass` (after `FUTURE = "future"`):

```python
    OPTION = "option"
```

And add near `Market` (a frozen dataclass alongside the others):

```python
@dataclass(frozen=True, slots=True)
class OptionContract:
    """One listed option, as returned by the OPRA chain. `right` is 'P' or 'C'; `osi_symbol` is the raw
    OSI symbol used to fetch that contract's daily bars."""
    osi_symbol: str
    expiry: "dt.date"
    strike: float
    right: str
```

(Ensure `from dataclasses import dataclass` and `import datetime as dt` are present in `models.py`.)

- [ ] **Step 4: Add the pure parsers + fetch methods to `databento.py`**

Add module-level parsers (pure, unit-tested above):

```python
def _chain_from_definition_df(df: "pd.DataFrame") -> list["OptionContract"]:
    """OPRA `definition` df -> list[OptionContract]. `instrument_class` 'P'/'C'; strike/expiry pretty-scaled
    by to_df()."""
    from fxhnt.domain.models import OptionContract
    out: list[OptionContract] = []
    for row in df.itertuples(index=False):
        cls = str(getattr(row, "instrument_class"))
        right = "P" if cls.upper().startswith("P") else "C"
        out.append(OptionContract(
            osi_symbol=str(getattr(row, "raw_symbol")),
            expiry=getattr(row, "expiration").date(),
            strike=float(getattr(row, "strike_price")),
            right=right,
        ))
    return out


def _bars_from_ohlcv_df(df: "pd.DataFrame") -> dict[str, dict[str, float]]:
    """OPRA `ohlcv-1d` df (index=ts, col `symbol`, col `close`) -> {osi_symbol: {iso_date: close}}."""
    out: dict[str, dict[str, float]] = {}
    for idx, sym, close in zip(df.index, df["symbol"], df["close"], strict=True):
        out.setdefault(str(sym), {})[str(idx.date())] = float(close)
    return out
```

Extend `_resolve` (add before the final `raise`):

```python
        if ac == AssetClass.OPTION:
            return self._option_dataset, market.symbol, "raw_symbol"
```

Add `option_dataset: str = "OPRA.PILLAR"` to `__init__` params and `self._option_dataset = option_dataset`.

Add the two methods on `DatabentoDataProvider`:

```python
    def fetch_option_chain(self, underlying_parent: str, as_of: str) -> list["OptionContract"]:
        """Listed XSP contracts on `as_of` via the OPRA `definition` schema (parent symbology, e.g.
        'XSP.OPT'). Cost-guarded."""
        import databento as db
        client = db.Historical()
        end = (dt.date.fromisoformat(as_of) + dt.timedelta(days=1)).isoformat()
        cost = client.metadata.get_cost(dataset=self._option_dataset, symbols=[underlying_parent],
                                        schema="definition", start=as_of, end=end, stype_in="parent")
        if cost > self._max_cost:
            raise DatabentoCostExceeded(
                f"estimated ${cost:.4f} > cap ${self._max_cost:.2f} for chain {underlying_parent}@{as_of}")
        data = client.timeseries.get_range(dataset=self._option_dataset, symbols=[underlying_parent],
                                           schema="definition", start=as_of, end=end, stype_in="parent")
        return _chain_from_definition_df(data.to_df())

    def fetch_option_bars(self, osi_symbols: list[str], start: str, end: str) -> dict[str, dict[str, float]]:
        """Daily closes for specific OPRA contracts (raw OSI symbols). Cost-guarded; empty list -> {}."""
        if not osi_symbols:
            return {}
        import databento as db
        client = db.Historical()
        cost = client.metadata.get_cost(dataset=self._option_dataset, symbols=osi_symbols,
                                        schema="ohlcv-1d", start=start, end=end, stype_in="raw_symbol")
        if cost > self._max_cost:
            raise DatabentoCostExceeded(
                f"estimated ${cost:.4f} > cap ${self._max_cost:.2f} for {len(osi_symbols)} option bars")
        data = client.timeseries.get_range(dataset=self._option_dataset, symbols=osi_symbols,
                                           schema="ohlcv-1d", start=start, end=end, stype_in="raw_symbol")
        return _bars_from_ohlcv_df(data.to_df())
```

- [ ] **Step 5: Run tests to verify pass**

Run: `.venv/bin/python -m pytest tests/unit/test_databento_options.py -q`
Expected: PASS (2 passed).

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/domain/models.py src/fxhnt/adapters/data/databento.py tests/unit/test_databento_options.py
git commit -m "feat(vrp): OPRA options path on the Databento adapter (chain + bars, cost-guarded)"
```

---

## Task 3: `xsp_option_bars` PIT table + repo

**Files:**
- Modify: `src/fxhnt/adapters/persistence/cockpit_models.py` (add `XspOptionBarRow`)
- Create: `src/fxhnt/adapters/persistence/xsp_option_bars.py`
- Test: `tests/unit/test_xsp_option_bars_repo.py`

**Interfaces:**
- Consumes: `OptionContract` (Task 2), `CockpitBase` (existing).
- Produces: `XspOptionBarsRepo(dsn)` with `migrate()`, `upsert_bars(rows: list[dict], at: dt.datetime)` (row keys: `osi_symbol, date(iso), expiry(iso), strike, right, close`), `bars_for(osi_symbols: list[str], start: str, end: str) -> dict[str, dict[str, float]]`, `chain_asof(date: str, dte_lo: int, dte_hi: int) -> list[OptionContract]` (contracts whose bar exists on `date` with `expiry ∈ [date+dte_lo, date+dte_hi]`).

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

```python
# tests/unit/test_xsp_option_bars_repo.py
import datetime as dt
from fxhnt.adapters.persistence.xsp_option_bars import XspOptionBarsRepo


def _repo(tmp_path):
    r = XspOptionBarsRepo(f"sqlite:///{tmp_path/'o.db'}")
    r.migrate()
    return r


def _bar(osi, date, expiry, strike, right, close):
    return dict(osi_symbol=osi, date=date, expiry=expiry, strike=strike, right=right, close=close)


def test_upsert_and_bars_for(tmp_path):
    r = _repo(tmp_path)
    at = dt.datetime(2024, 7, 15)
    r.upsert_bars([_bar("XSP240816P00470000", "2024-07-15", "2024-08-16", 470.0, "P", 1.20),
                   _bar("XSP240816P00470000", "2024-07-16", "2024-08-16", 470.0, "P", 1.35)], at)
    assert r.bars_for(["XSP240816P00470000"], "2024-07-15", "2024-07-16") == {
        "XSP240816P00470000": {"2024-07-15": 1.20, "2024-07-16": 1.35}}


def test_upsert_is_idempotent(tmp_path):
    r = _repo(tmp_path)
    at = dt.datetime(2024, 7, 15)
    r.upsert_bars([_bar("A", "2024-07-15", "2024-08-16", 470.0, "P", 1.20)], at)
    r.upsert_bars([_bar("A", "2024-07-15", "2024-08-16", 470.0, "P", 1.25)], at)  # same PK -> overwrite
    assert r.bars_for(["A"], "2024-07-15", "2024-07-15") == {"A": {"2024-07-15": 1.25}}


def test_chain_asof_filters_by_dte_window(tmp_path):
    r = _repo(tmp_path)
    at = dt.datetime(2024, 7, 15)
    r.upsert_bars([
        _bar("P30", "2024-07-15", "2024-08-14", 470.0, "P", 1.2),   # 30 DTE -> in [20,45]
        _bar("P10", "2024-07-15", "2024-07-25", 470.0, "P", 0.4),   # 10 DTE -> out
        _bar("P60", "2024-07-15", "2024-09-13", 470.0, "P", 2.0),   # 60 DTE -> out
    ], at)
    got = {c.osi_symbol for c in r.chain_asof("2024-07-15", 20, 45)}
    assert got == {"P30"}
```

- [ ] **Step 2: Run to verify failure**

Run: `.venv/bin/python -m pytest tests/unit/test_xsp_option_bars_repo.py -q`
Expected: FAIL (`ModuleNotFoundError`).

- [ ] **Step 3: Add the table** to `cockpit_models.py` (mirror `ForwardNavRow` style):

```python
class XspOptionBarRow(CockpitBase):
    __tablename__ = "xsp_option_bars"
    osi_symbol: Mapped[str] = mapped_column(String(32), primary_key=True)
    date: Mapped[dt.date] = mapped_column(Date, primary_key=True)
    expiry: Mapped[dt.date] = mapped_column(Date)
    strike: Mapped[float] = mapped_column(Float)
    right: Mapped[str] = mapped_column(String(1))
    close: Mapped[float] = mapped_column(Float)
    src_updated_at: Mapped[dt.datetime] = mapped_column(DateTime)
```

- [ ] **Step 4: Implement the repo**

```python
# src/fxhnt/adapters/persistence/xsp_option_bars.py
"""PIT store for frozen XSP option daily bars — the deterministic substrate the VRP recompute reads. OPRA is
hit only when FILLING this table (asset/backfill); VrpStrategy never calls OPRA. Same freeze discipline as the
equity Yahoo-PIT snapshot."""
from __future__ import annotations

import datetime as dt

from sqlalchemy import Date, create_engine, select
from sqlalchemy.orm import Session

from fxhnt.adapters.persistence.cockpit_models import CockpitBase, XspOptionBarRow
from fxhnt.domain.models import OptionContract


class XspOptionBarsRepo:
    def __init__(self, dsn: str) -> None:
        kw = {} if dsn.startswith("postgresql") else {"connect_args": {"check_same_thread": False}}
        self._engine = create_engine(dsn, **kw)

    def migrate(self) -> None:
        CockpitBase.metadata.create_all(self._engine)

    def upsert_bars(self, rows: list[dict], at: dt.datetime) -> None:
        with Session(self._engine) as s:
            for r in rows:
                obj = XspOptionBarRow(
                    osi_symbol=r["osi_symbol"], date=dt.date.fromisoformat(r["date"]),
                    expiry=dt.date.fromisoformat(r["expiry"]), strike=float(r["strike"]),
                    right=r["right"], close=float(r["close"]), src_updated_at=at)
                s.merge(obj)               # merge = insert-or-overwrite on the (osi_symbol, date) PK
            s.commit()

    def bars_for(self, osi_symbols: list[str], start: str, end: str) -> dict[str, dict[str, float]]:
        if not osi_symbols:
            return {}
        lo, hi = dt.date.fromisoformat(start), dt.date.fromisoformat(end)
        out: dict[str, dict[str, float]] = {}
        with Session(self._engine) as s:
            q = select(XspOptionBarRow).where(
                XspOptionBarRow.osi_symbol.in_(osi_symbols),
                XspOptionBarRow.date >= lo, XspOptionBarRow.date <= hi)
            for row in s.scalars(q):
                out.setdefault(row.osi_symbol, {})[row.date.isoformat()] = row.close
        return out

    def chain_asof(self, date: str, dte_lo: int, dte_hi: int) -> list[OptionContract]:
        d = dt.date.fromisoformat(date)
        exp_lo, exp_hi = d + dt.timedelta(days=dte_lo), d + dt.timedelta(days=dte_hi)
        with Session(self._engine) as s:
            q = select(XspOptionBarRow).where(
                XspOptionBarRow.date == d, XspOptionBarRow.right == "P",
                XspOptionBarRow.expiry >= exp_lo, XspOptionBarRow.expiry <= exp_hi)
            return [OptionContract(osi_symbol=r.osi_symbol, expiry=r.expiry, strike=r.strike, right=r.right)
                    for r in s.scalars(q)]
```

- [ ] **Step 5: Run tests to verify pass**

Run: `.venv/bin/python -m pytest tests/unit/test_xsp_option_bars_repo.py -q`
Expected: PASS (3 passed).

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/adapters/persistence/cockpit_models.py src/fxhnt/adapters/persistence/xsp_option_bars.py tests/unit/test_xsp_option_bars_repo.py
git commit -m "feat(vrp): xsp_option_bars PIT table + repo (upsert/bars_for/chain_asof)"
```

---

## Task 4: `VrpStrategy` — defined-risk ladder recomputed off frozen bars

**Files:**
- Create: `src/fxhnt/application/vrp_book.py`
- Test: `tests/unit/test_vrp_book.py`

**Interfaces:**
- Consumes: `vrp_pricing` (Task 1), `XspOptionBarsRepo` (Task 3), `OptionContract` (Task 2), `allocation_engine.cvar_daily` (merged).
- Produces: `VrpStrategy(bars: XspOptionBarsRepo, *, dte_lo=20, dte_hi=45, dte_target=30, short_delta=0.20, width_points=5.0, ladder=5, profit_take=0.5, entry_weekday=0)` implementing `advance(last_date: str | None, extra: dict) -> tuple[list[tuple[str, float]], dict]`. `margin = width_points * 100`.

**Design (recompute mode):** `extra` carries `open_spreads` (list of dicts) and `last_day`. On `advance`, gather frozen trading days after `last_date` (union of dates present in the table, via a light `bars.trading_days(after)` helper — add it in this task). For each day in order: (1) mark each open spread's two legs on that day's closes, book `sum(daily_mtm_change) / (n_open * margin)`; (2) close spreads hitting 50%-profit or DTE≤1; (3) on `entry_weekday`, if `len(open) < ladder`, select a new spread from `chain_asof`. `advance(None, {})` walks the entire frozen history from the first available day.

- [ ] **Step 1: Write the failing tests** (synthetic frozen bars → deterministic series + defined-risk bounds)

```python
# tests/unit/test_vrp_book.py
import datetime as dt
import math
from fxhnt.adapters.persistence.xsp_option_bars import XspOptionBarsRepo
from fxhnt.application.vrp_book import VrpStrategy
from fxhnt.application.allocation_engine import cvar_daily


def _seed(tmp_path, days, contracts):
    """contracts: list of (osi, expiry, strike, right, {iso_date: close}). Returns a repo."""
    r = XspOptionBarsRepo(f"sqlite:///{tmp_path/'v.db'}")
    r.migrate()
    at = dt.datetime(2024, 1, 1)
    rows = []
    for osi, expiry, strike, right, closes in contracts:
        for d, c in closes.items():
            rows.append(dict(osi_symbol=osi, date=d, expiry=expiry, strike=strike, right=right, close=c))
    r.upsert_bars(rows, at)
    return r


def _straight_chain(start="2024-07-01", n=10, expiry="2024-07-31", forward=470.0):
    """A flat-vol synthetic put chain across strikes, plus an ATM call for the parity forward."""
    days = [(dt.date.fromisoformat(start) + dt.timedelta(days=i)).isoformat() for i in range(n)]
    contracts = []
    for k in range(440, 475, 5):                       # put strikes below/at forward
        osi = f"XSP{expiry.replace('-','')[2:]}P{k*1000:08d}"
        closes = {d: max(0.05, (470 - k) * 0.02 + 0.8) for d in days}   # calm decay-ish marks
        contracts.append((osi, expiry, float(k), "P", closes))
    # ATM call so parity_forward(K=470) recovers ~470
    contracts.append((f"XSP{expiry.replace('-','')[2:]}C{470*1000:08d}", expiry, 470.0, "C",
                      {d: 0.8 for d in days}))
    return days, contracts


def test_advance_full_history_is_deterministic(tmp_path):
    days, contracts = _straight_chain()
    r = _seed(tmp_path, days, contracts)
    a, _ = VrpStrategy(r, entry_weekday=0).advance(None, {})
    b, _ = VrpStrategy(r, entry_weekday=0).advance(None, {})
    assert a == b                                      # same frozen input -> identical series
    assert all(isinstance(d, str) and isinstance(x, float) for d, x in a)


def test_defined_risk_daily_loss_is_bounded(tmp_path):
    # a crash day: the short put mark spikes; the spread loss per day cannot exceed width/margin = 1.0
    days, contracts = _straight_chain(n=6)
    # overwrite the last day's marks with a crash spike on all puts
    crash = days[-1]
    for osi, expiry, strike, right, closes in contracts:
        if right == "P":
            closes[crash] = closes[crash] + 50.0
    r = _seed(tmp_path, days, contracts)
    series, _ = VrpStrategy(r, entry_weekday=0).advance(None, {})
    assert series, "series should not be empty"
    assert min(x for _, x in series) >= -1.0 - 1e-9    # defined-risk: daily loss per margin bounded
    assert max(abs(x) for _, x in series) < 5.0        # sane magnitude


def test_fat_left_tail_triggers_cvar_floor(tmp_path):
    # a VRP-shaped series (small positive drift + occasional deep loss) is fatter-left than Gaussian
    series = [0.002] * 40 + [-0.08, -0.06]
    ratio = cvar_daily(series, 0.10) / (sum((x - sum(series)/len(series))**2 for x in series) / len(series))**0.5
    assert ratio > 1.755                               # fatter than Gaussian ES-10% -> CVaR floor bites
```

- [ ] **Step 2: Run to verify failure**

Run: `.venv/bin/python -m pytest tests/unit/test_vrp_book.py -q`
Expected: FAIL (`ModuleNotFoundError: fxhnt.application.vrp_book`).

- [ ] **Step 3: Add a `trading_days` helper to the repo** (`xsp_option_bars.py`, needed by the strategy):

```python
    def trading_days(self, after: str | None) -> list[str]:
        """Ascending distinct dates present in the table, strictly after `after` (or all if None)."""
        from sqlalchemy import distinct
        with Session(self._engine) as s:
            q = select(distinct(XspOptionBarRow.date)).order_by(XspOptionBarRow.date)
            days = [d.isoformat() for d in s.scalars(q)]
        return [d for d in days if after is None or d > after]
```

- [ ] **Step 4: Implement `VrpStrategy`**

```python
# src/fxhnt/application/vrp_book.py
"""The defined-risk XSP put-credit-spread VRP sleeve (recompute mode). Recomputes the daily book ENTIRELY off
frozen `xsp_option_bars` rows — put-call-parity forward (Task 1) -> IV inversion -> ~20Δ short strike, long a
fixed width below -> real-mark daily P&L per dollar of defined-risk margin. Never calls OPRA. Mirrors how
MultiStratStrategy produces both the full-history ref (advance(None,{})) and the nightly forward window."""
from __future__ import annotations

import datetime as dt
from typing import Any

from fxhnt.adapters.persistence.xsp_option_bars import XspOptionBarsRepo
from fxhnt.application import vrp_pricing


class VrpStrategy:
    def __init__(self, bars: XspOptionBarsRepo, *, dte_lo: int = 20, dte_hi: int = 45, dte_target: int = 30,
                 short_delta: float = 0.20, width_points: float = 5.0, ladder: int = 5,
                 profit_take: float = 0.5, entry_weekday: int = 0) -> None:
        self._bars = bars
        self._dte_lo, self._dte_hi, self._dte_target = dte_lo, dte_hi, dte_target
        self._short_delta, self._width, self._ladder = short_delta, width_points, ladder
        self._profit_take, self._entry_weekday = profit_take, entry_weekday
        self.margin = width_points * 100.0

    @staticmethod
    def _dte(day: str, expiry: dt.date) -> int:
        return (expiry - dt.date.fromisoformat(day)).days

    def _select_spread(self, day: str) -> dict | None:
        """Pick the ~30-DTE, ~20Δ short put + long a width below, priced from frozen marks on `day`."""
        chain = self._bars.chain_asof(day, self._dte_lo, self._dte_hi)
        if not chain:
            return None
        # nearest expiry to dte_target
        exp = min({c.expiry for c in chain}, key=lambda e: abs(self._dte(day, e) - self._dte_target))
        puts = sorted((c for c in chain if c.expiry == exp), key=lambda c: c.strike)
        tau = max(self._dte(day, exp), 1) / 365.0
        marks = self._bars.bars_for([c.osi_symbol for c in puts], day, day)
        priced = [(c, marks.get(c.osi_symbol, {}).get(day)) for c in puts]
        priced = [(c, p) for c, p in priced if p is not None and p > 0.0]
        if len(priced) < 2:
            return None
        # forward from parity: use the ATM put + a matching call if present, else approximate F = highest strike
        atm = min(priced, key=lambda cp: abs(cp[0].strike - priced[-1][0].strike))
        forward = atm[0].strike  # ATM put strike ~ forward when call absent; refined by parity when call frozen
        # choose short strike whose |delta| is nearest short_delta
        best = None
        for c, price in priced:
            iv = vrp_pricing.implied_vol_put(price, forward, c.strike, tau)
            if iv is None:
                continue
            dlt = vrp_pricing.put_delta(forward, c.strike, iv, tau)
            score = abs(dlt - self._short_delta)
            if best is None or score < best[0]:
                best = (score, c, price)
        if best is None:
            return None
        _, short_c, short_price = best
        long_strike = short_c.strike - self._width
        long_c = next((c for c, _ in priced if abs(c.strike - long_strike) < 1e-6), None)
        long_price = self._bars.bars_for([long_c.osi_symbol], day, day).get(long_c.osi_symbol, {}).get(day) if long_c else None
        if long_c is None or long_price is None:
            return None
        credit = short_price - long_price
        if credit <= 0.0:
            return None
        return dict(entry=day, expiry=exp.isoformat(), short_osi=short_c.osi_symbol, long_osi=long_c.osi_symbol,
                    credit=credit)

    def _spread_value(self, sp: dict, day: str) -> float | None:
        s = self._bars.bars_for([sp["short_osi"]], day, day).get(sp["short_osi"], {}).get(day)
        l = self._bars.bars_for([sp["long_osi"]], day, day).get(sp["long_osi"], {}).get(day)
        if s is None or l is None:
            return None
        return s - l                                       # current cost to close the spread

    def advance(self, last_date: str | None, extra: dict[str, Any]) -> tuple[list[tuple[str, float]], dict[str, Any]]:
        opens: list[dict] = list(extra.get("open_spreads", []))
        rows: list[tuple[str, float]] = []
        for day in self._bars.trading_days(last_date):
            # 1) mark open spreads -> daily defined-risk return
            pnl = 0.0
            still: list[dict] = []
            for sp in opens:
                val = self._spread_value(sp, day)
                if val is None:
                    still.append(sp)
                    continue
                prev = sp.get("last_val", sp["credit"])    # entry day: mark vs credit
                pnl += (prev - val)                        # short-vol: value falling = profit
                sp["last_val"] = val
                # 2) management: 50% profit or expiry-1
                if val <= (1.0 - self._profit_take) * sp["credit"] or self._dte(day, dt.date.fromisoformat(sp["expiry"])) <= 1:
                    continue                               # closed (drop from book)
                still.append(sp)
            opens = still
            n_open = max(1, len(opens))
            rows.append((day, pnl / (n_open * self.margin)))
            # 3) weekly entry
            if dt.date.fromisoformat(day).weekday() == self._entry_weekday and len(opens) < self._ladder:
                sp = self._select_spread(day)
                if sp is not None:
                    sp["last_val"] = sp["credit"]
                    opens.append(sp)
        return rows, {"open_spreads": opens}
```

- [ ] **Step 5: Run tests to verify pass**

Run: `.venv/bin/python -m pytest tests/unit/test_vrp_book.py -q`
Expected: PASS (3 passed). If the defined-risk bound test fails, the daily-return normalization is wrong — the per-day loss must divide by `n_open * margin`, never book raw dollars.

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/application/vrp_book.py src/fxhnt/adapters/persistence/xsp_option_bars.py tests/unit/test_vrp_book.py
git commit -m "feat(vrp): VrpStrategy defined-risk XSP put-spread ladder (recompute off frozen bars)"
```

---

## Task 5: Registry `vrp` entry + migration anchor

**Files:**
- Modify: `src/fxhnt/registry.py` (add `"vrp"` to `STRATEGY_REGISTRY`)
- Modify: `src/fxhnt/adapters/orchestration/migration_builders.py` (add `builders["vrp"]`)
- Test: `tests/unit/test_registry_vrp.py`

**Interfaces:**
- Consumes: `VrpStrategy` (Task 4), `XspOptionBarsRepo` (Task 3).
- Produces: registry id `vrp`; `builders["vrp"] = ("vrp_state", <build VrpStrategy>)`.

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

```python
# tests/unit/test_registry_vrp.py
from fxhnt.registry import STRATEGY_REGISTRY


def test_vrp_registry_entry_shape():
    e = STRATEGY_REGISTRY["vrp"]
    assert e["venue"] == "ibkr-xsp"
    assert e["sleeve"] == "equity-vol"
    assert e["tier"] == "research"
    assert e["execution"] == "paper"
    assert e["state_file"] == "vrp_state"
    assert e["gate_spec"] == {"gate_type": "reconciliation", "min_forward_days": 21}
    assert e["definition"]["record_mode"] == "recompute"
    assert e["promote_on_pass"] is False
```

- [ ] **Step 2: Run to verify failure**

Run: `.venv/bin/python -m pytest tests/unit/test_registry_vrp.py -q`
Expected: FAIL (`KeyError: 'vrp'`).

- [ ] **Step 3: Add the registry entry** (in `STRATEGY_REGISTRY`, after `stablecoin_rotation`):

```python
    # Defined-risk equity VRP (XSP put-credit-spreads) on REAL OPRA option data — the cross-asset diversifier
    # (co-crashes systemic risk-off, so NOT a systemic hedge; diversifies idiosyncratic crypto crashes).
    # record_mode recompute off the frozen xsp_option_bars PIT table. promote_on_pass False: real-capital-
    # adjacent, so promotion to deploy (CVaR-sized capital) is a deliberate human step after it reconciles.
    "vrp": {
        "display_name": "Equity VRP (XSP put-credit-spreads, defined-risk)", "sleeve": "equity-vol",
        "venue": "ibkr-xsp", "state_file": "vrp_state",
        "tier": "research", "execution": "paper", "promote_on_pass": False,
        "gate_spec": {"gate_type": "reconciliation", "min_forward_days": 21},
        "definition": {"params": {"instrument": "XSP", "structure": "put-credit-spread",
                                  "dte_target": 30, "short_delta": 0.20, "width_points": 5.0,
                                  "ladder": 5, "source": "opra-pit"},
                       "version": 1, "record_mode": "recompute"},
    },
```

- [ ] **Step 4: Add the migration builder** (in `migration_builders.py`, after `_build_multistrat`):

```python
    # ---- vrp_nav: defined-risk XSP put-spread VRP on the frozen OPRA PIT table (recompute) ----------------
    def _build_vrp() -> Any:
        from fxhnt.adapters.persistence.xsp_option_bars import XspOptionBarsRepo
        from fxhnt.application.vrp_book import VrpStrategy

        return VrpStrategy(XspOptionBarsRepo(settings.operational_dsn))

    builders["vrp"] = ("vrp_state", _build_vrp)
```

- [ ] **Step 5: Run tests to verify pass**

Run: `.venv/bin/python -m pytest tests/unit/test_registry_vrp.py tests/unit/test_registry_tiers.py -q`
Expected: PASS (existing registry tests still green).

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/registry.py src/fxhnt/adapters/orchestration/migration_builders.py tests/unit/test_registry_vrp.py
git commit -m "feat(vrp): registry vrp entry + migration anchor builder"
```

---

## Task 6: `vrp_nav` asset + definitions wiring

**Files:**
- Modify: `src/fxhnt/adapters/orchestration/assets.py` (add `vrp_nav` + a `_freeze_xsp_slice` helper)
- Modify: `src/fxhnt/adapters/orchestration/definitions.py` (add `vrp_nav` to the job + cockpit_forward deps)
- Test: `tests/integration/test_vrp_nav_asset.py`

**Interfaces:**
- Consumes: `DatabentoDataProvider` (Task 2), `XspOptionBarsRepo` (Task 3), `VrpStrategy` (Task 4), `_run_paper_tracker` + `_persist_track_backtest_ref` (existing), `run_track` (existing).
- Produces: `vrp_nav` asset writing `forward_summary["vrp"]` + `backtest_summary["vrp"]`.

**Freeze policy (Global Constraints):** each run, freeze the daily slice — all XSP puts with DTE ∈ [20,45] and moneyness strike/forward ∈ [0.80,1.02] for `today` — into `xsp_option_bars`, plus the held-ladder legs' marks, then `run_track(repo, "vrp", VrpStrategy(bars), today, at)` with `persist_backtest_ref=True`.

- [ ] **Step 1: Write the failing integration test** (inject a fake data provider — no live OPRA)

```python
# tests/integration/test_vrp_nav_asset.py
import datetime as dt
from fxhnt.adapters.persistence.xsp_option_bars import XspOptionBarsRepo
from fxhnt.adapters.orchestration.assets import _freeze_xsp_slice


class _FakeDbn:
    def fetch_option_chain(self, parent, as_of):
        from fxhnt.domain.models import OptionContract
        return [OptionContract(f"XSP240816P00{k}000", dt.date(2024, 8, 16), float(k), "P")
                for k in range(440, 475, 5)]

    def fetch_option_bars(self, osi, start, end):
        return {s: {start: 1.0} for s in osi}


def test_freeze_xsp_slice_writes_dte_moneyness_band(tmp_path):
    bars = XspOptionBarsRepo(f"sqlite:///{tmp_path/'o.db'}")
    bars.migrate()
    _freeze_xsp_slice(_FakeDbn(), bars, as_of="2024-07-20", forward=470.0,
                      at=dt.datetime(2024, 7, 20))
    # 30-DTE puts within moneyness [0.80,1.02]*470 = [376,479.4] -> all seeded strikes qualify
    chain = bars.chain_asof("2024-07-20", 20, 45)
    assert {c.strike for c in chain} == {float(k) for k in range(440, 475, 5)}
```

- [ ] **Step 2: Run to verify failure**

Run: `.venv/bin/python -m pytest tests/integration/test_vrp_nav_asset.py -q`
Expected: FAIL (`ImportError: cannot import name '_freeze_xsp_slice'`).

- [ ] **Step 3: Add `_freeze_xsp_slice` + the `vrp_nav` asset** to `assets.py`

```python
def _freeze_xsp_slice(dbn, bars, *, as_of: str, forward: float, at) -> None:  # type: ignore[no-untyped-def]
    """Freeze the deterministic daily slice: XSP puts with DTE in [20,45] and strike/forward in [0.80,1.02]
    for `as_of`, into the PIT table. This is the ONLY place OPRA is read for VRP."""
    import datetime as _dt
    chain = dbn.fetch_option_chain("XSP.OPT", as_of)
    d = _dt.date.fromisoformat(as_of)
    keep = [c for c in chain if c.right == "P"
            and 20 <= (c.expiry - d).days <= 45
            and 0.80 <= c.strike / forward <= 1.02]
    marks = dbn.fetch_option_bars([c.osi_symbol for c in keep], as_of, as_of)
    rows = [dict(osi_symbol=c.osi_symbol, date=as_of, expiry=c.expiry.isoformat(),
                 strike=c.strike, right="P", close=marks[c.osi_symbol][as_of])
            for c in keep if c.osi_symbol in marks and as_of in marks[c.osi_symbol]]
    bars.upsert_bars(rows, at)


@asset
def vrp_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """Paper-forward defined-risk XSP put-credit-spread VRP on REAL OPRA data. Freezes today's DTE/moneyness
    put slice PIT into xsp_option_bars, then recomputes off the FROZEN table (spec §A) so the record is stable
    against any OPRA revision. persist_backtest_ref=True publishes the basis-matched real ref for the gate."""
    import datetime as _dt

    from fxhnt.adapters.data.databento import DatabentoDataProvider
    from fxhnt.adapters.persistence.xsp_option_bars import XspOptionBarsRepo
    from fxhnt.application.vrp_book import VrpStrategy
    from fxhnt.config import get_settings

    s = get_settings()
    bars = XspOptionBarsRepo(s.operational_dsn)
    bars.migrate()
    dbn = DatabentoDataProvider()
    at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)
    today = _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%d")
    fwd = _xsp_forward_estimate(dbn, today)                 # parity forward from the frozen ATM pair
    _freeze_xsp_slice(dbn, bars, as_of=today, forward=fwd, at=at)

    return _run_paper_tracker(
        context, "vrp_nav", lambda: VrpStrategy(bars), "vrp_state", persist_backtest_ref=True,
    )
```

Add `_xsp_forward_estimate` (parity forward from the ATM call/put on `today`; falls back to the SPY×10 proxy only if the ATM pair is missing):

```python
def _xsp_forward_estimate(dbn, as_of: str) -> float:  # type: ignore[no-untyped-def]
    """Implied XSP forward from put-call parity on the near-ATM pair of the ~30-DTE expiry."""
    from fxhnt.application.vrp_pricing import parity_forward
    chain = [c for c in dbn.fetch_option_chain("XSP.OPT", as_of) if 20 <= (c.expiry - _iso_date(as_of)).days <= 45]
    if not chain:
        raise RuntimeError(f"no XSP chain for {as_of} — cannot estimate forward")
    exp = min({c.expiry for c in chain}, key=lambda e: abs((e - _iso_date(as_of)).days - 30))
    puts = {c.strike: c for c in chain if c.expiry == exp and c.right == "P"}
    calls = {c.strike: c for c in chain if c.expiry == exp and c.right == "C"}
    common = sorted(set(puts) & set(calls))
    if not common:
        raise RuntimeError(f"no matched call/put strike for {as_of} forward")
    k = common[len(common) // 2]                            # mid strike ≈ ATM
    marks = dbn.fetch_option_bars([puts[k].osi_symbol, calls[k].osi_symbol], as_of, as_of)
    p = marks[puts[k].osi_symbol][as_of]
    c = marks[calls[k].osi_symbol][as_of]
    return parity_forward(call_atm=c, put_atm=p, k_atm=k)


def _iso_date(s: str):  # type: ignore[no-untyped-def]
    import datetime as _dt
    return _dt.date.fromisoformat(s)
```

- [ ] **Step 4: Wire into `definitions.py`** — add `vrp_nav` to `combined_book_forward_job.selection`, to `Definitions.assets`, and to `cockpit_forward`'s `deps`. (Match the existing `multistrat_nav` wiring line-for-line; grep `multistrat_nav` in `definitions.py` and add `vrp_nav` beside each occurrence.)

- [ ] **Step 5: Run tests to verify pass**

Run: `.venv/bin/python -m pytest tests/integration/test_vrp_nav_asset.py -q && .venv/bin/python -c "from fxhnt.adapters.orchestration.definitions import defs; print('defs load OK')"`
Expected: PASS + `defs load OK` (the Dagster `Definitions` still imports with `vrp_nav` wired).

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/adapters/orchestration/assets.py src/fxhnt/adapters/orchestration/definitions.py tests/integration/test_vrp_nav_asset.py
git commit -m "feat(vrp): vrp_nav asset (freeze OPRA PIT slice -> recompute + real backtest ref) + wiring"
```

---

## Task 7: VRP execution twin (IBKR XSP `mleg`) — coded, SUSPENDED

**Files:**
- Modify: `src/fxhnt/adapters/broker/ibkr.py` (add `place_combo`)
- Create: `src/fxhnt/application/vrp_exec_record.py`
- Modify: `src/fxhnt/cli.py` (add `execute-vrp`)
- Create: `infra/k8s/jobs/fxhnt-vrp-rebalancer.yaml` (suspended weekly) + `infra/k8s/jobs/fxhnt-opra-backfill.yaml`
- Modify: `src/fxhnt/registry.py` (add `vrp_exec` observed twin)
- Test: `tests/unit/test_vrp_exec_record.py`, `tests/unit/test_registry_vrp.py` (extend)

**Interfaces:**
- Consumes: `IbkrBroker` (existing), `ForwardNavRepo` (existing), `_exec_venue_string` / `_resolve_paper_envelope` (existing in cli.py), `DatabentoDataProvider.fetch_option_chain/bars` (Task 2), `VrpStrategy._select_spread` (Task 4).
- Produces: `IbkrBroker.place_combo(legs: list[tuple[str, str, int]], net_limit: float) -> str` (legs = `(osi_symbol, "BUY"|"SELL", qty)`, `net_limit` negative = credit); `plan_and_record_vrp(...)`; registry id `vrp_exec` (`venue` `"ibkr-xsp"`, `record_mode` `"observed"`).

- [ ] **Step 1: Write the failing tests** (paper-envelope refusal on non-paper; combo leg construction)

```python
# tests/unit/test_vrp_exec_record.py
import pytest
from fxhnt.application.vrp_exec_record import build_combo_legs


def test_build_combo_legs_is_defined_risk_credit():
    legs, net = build_combo_legs(short_osi="XSP..P00470000", long_osi="XSP..P00465000",
                                 short_price=1.20, long_price=0.40, qty=1)
    assert ("XSP..P00470000", "SELL", 1) in legs
    assert ("XSP..P00465000", "BUY", 1) in legs
    assert net == pytest.approx(-0.80)              # negative limit = net credit received
```

```python
# tests/unit/test_registry_vrp.py  (append)
def test_vrp_exec_twin_is_observed():
    from fxhnt.registry import STRATEGY_REGISTRY
    e = STRATEGY_REGISTRY["vrp_exec"]
    assert e["venue"] == "ibkr-xsp"
    assert e["definition"]["record_mode"] == "observed"
    assert e["gate_spec"]["gate_type"] == "reconciliation"
```

- [ ] **Step 2: Run to verify failure**

Run: `.venv/bin/python -m pytest tests/unit/test_vrp_exec_record.py tests/unit/test_registry_vrp.py -q`
Expected: FAIL (module + KeyError).

- [ ] **Step 3: Add `build_combo_legs` + `plan_and_record_vrp`** (`vrp_exec_record.py`)

```python
# src/fxhnt/application/vrp_exec_record.py
"""Executed VRP twin (sub-project C-style): place the atomic XSP put-credit-spread on IBKR and record the
executed-vs-modeled reconciliation row. SUSPENDED until the account has options permission. build_combo_legs
is pure + tested; the record path mirrors multistrat_exec_record."""
from __future__ import annotations


def build_combo_legs(*, short_osi: str, long_osi: str, short_price: float, long_price: float,
                     qty: int) -> tuple[list[tuple[str, str, int]], float]:
    """Defined-risk put-credit-spread: SELL the short put, BUY the wing. Net limit is negative (a credit)."""
    legs = [(short_osi, "SELL", qty), (long_osi, "BUY", qty)]
    net = -(short_price - long_price)                       # credit received -> negative limit
    return legs, round(net, 2)
```

`plan_and_record_vrp(repo, *, dbn, broker, nlv, today, at, paper_envelope, account_is_paper, venue, execute)`: refuse (`ValueError("paper-envelope refused …")`, zero orders) if `paper_envelope > 0 and not account_is_paper`; select the spread via a `VrpStrategy(...)._select_spread`-equivalent on the live chain; `build_combo_legs`; if `execute`, `broker.place_combo(legs, net)`; record the observed `vrp_exec` return row (envelope-basis P&L) via `repo`. (Mirror `multistrat_exec_record.plan_and_record` guard-for-guard; a unit test covers the refusal.)

- [ ] **Step 4: Add `place_combo` to `IbkrBroker`**

```python
    def place_combo(self, legs: list[tuple[str, str, int]], net_limit: float) -> str:
        """Atomic XSP put-credit-spread as a BAG combo. legs = (osi_symbol, side, qty); net_limit<0 = credit.
        Requires options trading permission (account currently lacks it -> this path ships suspended)."""
        from ib_async import Bag, ComboLeg, LimitOrder, Option
        combo_legs = []
        for osi, side, qty in legs:
            opt = Option()  # qualified from the OSI symbol
            opt.localSymbol = osi
            opt.exchange = "SMART"
            cid = self._ib.qualifyContracts(opt)[0].conId
            combo_legs.append(ComboLeg(conId=cid, ratio=qty, action=side, exchange="SMART"))
        bag = Bag(symbol="XSP", exchange="SMART", currency="USD", comboLegs=combo_legs)
        total_qty = legs[0][2]
        order = LimitOrder("SELL", total_qty, net_limit)   # SELL the spread for a net credit
        trade = self._ib.placeOrder(bag, order)
        self._ib.sleep(1)
        return str(trade.order.orderId)
```

- [ ] **Step 5: Add the `execute-vrp` CLI command + `vrp_exec` registry twin + suspended CronJob/backfill YAML.** `execute-vrp` mirrors `execute-multistrat`'s structure (IBKR broker context, `_resolve_paper_envelope`, `_exec_venue_string`, `plan_and_record_vrp`). Registry `vrp_exec` (append after `vrp`):

```python
    "vrp_exec": {
        "display_name": "Equity VRP (EXECUTED, IBKR XSP)", "sleeve": "equity-vol",
        "venue": "ibkr-xsp", "state_file": "vrp_exec_state",
        "tier": "research", "execution": "paper",
        "gate_spec": {"gate_type": "reconciliation", "min_forward_days": 21},
        "definition": {"params": {"source": "ibkr-xsp-fills", "models": "vrp"},
                       "version": 1, "record_mode": "observed"},
    },
```

`infra/k8s/jobs/fxhnt-vrp-rebalancer.yaml`: copy `fxhnt-ibkr-rebalancer.yaml`; change name→`fxhnt-vrp-rebalancer`, `schedule: "40 14 * * 1"` (weekly Monday), **`suspend: true`**, command→`pip install --quiet --no-cache-dir ib_async==2.1.0 databento==0.79.0 && fxhnt execute-vrp --execute --paper-envelope 1000000`, add `DATABENTO_API_KEY` from `databento-credentials`, extend egress to OPRA/Databento 443 (already `0.0.0.0/0:443`). `infra/k8s/jobs/fxhnt-opra-backfill.yaml`: a one-time `Job` (not CronJob) running the 2013→ backfill under a raised cap, chunked by month (`fxhnt backfill-xsp-opra --start 2013-04-01 --max-cost 5.0`), also suspended-by-default (apply manually).

- [ ] **Step 6: Run tests to verify pass**

Run: `.venv/bin/python -m pytest tests/unit/test_vrp_exec_record.py tests/unit/test_registry_vrp.py -q && .venv/bin/python -c "import yaml; yaml.safe_load_all(open('infra/k8s/jobs/fxhnt-vrp-rebalancer.yaml')); print('yaml OK')"`
Expected: PASS + `yaml OK`; confirm `suspend: true` in the CronJob.

- [ ] **Step 7: Commit**

```bash
git add src/fxhnt/adapters/broker/ibkr.py src/fxhnt/application/vrp_exec_record.py src/fxhnt/cli.py src/fxhnt/registry.py infra/k8s/jobs/fxhnt-vrp-rebalancer.yaml infra/k8s/jobs/fxhnt-opra-backfill.yaml tests/unit/test_vrp_exec_record.py tests/unit/test_registry_vrp.py
git commit -m "feat(vrp): IBKR XSP mleg combo + execute-vrp + vrp_exec twin (SUSPENDED)"
```

---

## Task 8: Component B — UCITS execution routing (thin), SUSPENDED

**Files:**
- Modify: `src/fxhnt/config.py` (add `UcitsSettings`)
- Create: `src/fxhnt/application/ucits_map.py`
- Modify: `src/fxhnt/cli.py:257-324` (`execute-multistrat` `--venue us|ucits`)
- Create: `infra/k8s/jobs/fxhnt-ucits-rebalancer.yaml` (suspended monthly)
- Test: `tests/unit/test_ucits_routing.py`

**Interfaces:**
- Consumes: `execute-multistrat` weights/prices (existing), `_exec_venue_string` (existing).
- Produces: `apply_ucits_map(weights: dict[str,float], prices: dict[str,float], mapping: dict[str,str]) -> tuple[dict, dict]` (returns UCITS-keyed weights+prices; raises `ValueError` on an unmapped sleeve); `UcitsSettings.map: dict[str,str]`.

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

```python
# tests/unit/test_ucits_routing.py
import pytest
from fxhnt.application.ucits_map import apply_ucits_map

MAP = {"SPY": "CSPX", "IEF": "IBTM", "GLD": "SGLN", "PDBC": "ICOM", "DBMF": "DBMF.L"}


def test_apply_ucits_map_rewrites_symbols():
    w, p = apply_ucits_map({"SPY": 0.6, "GLD": 0.4}, {"SPY": 600.0, "GLD": 200.0}, MAP)
    assert w == {"CSPX": 0.6, "SGLN": 0.4}
    assert p == {"CSPX": 600.0, "SGLN": 200.0}


def test_unmapped_sleeve_refuses():
    with pytest.raises(ValueError, match="unmapped"):
        apply_ucits_map({"SPY": 0.5, "QQQ": 0.5}, {"SPY": 1.0, "QQQ": 1.0}, MAP)
```

- [ ] **Step 2: Run to verify failure**

Run: `.venv/bin/python -m pytest tests/unit/test_ucits_routing.py -q`
Expected: FAIL (`ModuleNotFoundError`).

- [ ] **Step 3: Implement the map + config**

```python
# src/fxhnt/application/ucits_map.py
"""US->UCITS execution remap for the PRIIPs-legal multistrat leg. The MODELED track is unchanged (models US
underlyings); this rewrites ONLY the execute path's symbols. An unmapped sleeve refuses — never route an
order we can't map."""
from __future__ import annotations


def apply_ucits_map(weights: dict[str, float], prices: dict[str, float],
                    mapping: dict[str, str]) -> tuple[dict[str, float], dict[str, float]]:
    missing = [s for s in weights if s not in mapping]
    if missing:
        raise ValueError(f"unmapped sleeve(s) for UCITS routing: {missing} — refusing to route")
    w = {mapping[s]: v for s, v in weights.items()}
    p = {mapping[s]: prices[s] for s in weights}
    return w, p
```

Add to `config.py` (a pydantic settings model + field on `Settings`):

```python
class UcitsSettings(BaseModel):
    # US-domiciled ticker -> Euronext UCITS equivalent (execute-only; modeled track stays US).
    map: dict[str, str] = Field(default_factory=lambda: {
        "SPY": "CSPX", "IEF": "IBTM", "GLD": "SGLN", "PDBC": "ICOM", "DBMF": "DBMF.L"})
```

and `ucits: UcitsSettings = Field(default_factory=UcitsSettings)` on `Settings` (env prefix gives `FXHNT_UCITS_MAP`).

- [ ] **Step 4: Wire `--venue` into `execute-multistrat`** (`cli.py`). Add the option and apply the map before building the broker order:

```python
    venue_kind: str = typer.Option("us", "--venue", help="'us' (US ETFs, default) or 'ucits' (Euronext UCITS)"),
```

After `weights` is computed and `bweights/bprices` are set, insert:

```python
    if venue_kind == "ucits":
        from fxhnt.application.ucits_map import apply_ucits_map
        bweights, bprices = apply_ucits_map(bweights, bprices, settings.ucits.map)
        typer.echo("UCITS routing: " + ", ".join(f"{s} {w:+.3f}" for s, w in bweights.items()))
```

and thread the recorded venue: when `venue_kind == "ucits"`, override `venue = "ibkr-ucits"` after `_exec_venue_string(...)`.

- [ ] **Step 5: Add the suspended CronJob** `infra/k8s/jobs/fxhnt-ucits-rebalancer.yaml`: copy `fxhnt-ibkr-rebalancer.yaml`; name→`fxhnt-ucits-rebalancer`, `schedule: "40 14 1 * *"` (monthly, 1st), **`suspend: true`**, command→`... && fxhnt execute-multistrat --broker ibkr --venue ucits --execute --paper-envelope 1000000`.

- [ ] **Step 6: Run tests to verify pass**

Run: `.venv/bin/python -m pytest tests/unit/test_ucits_routing.py tests/unit/test_execute_multistrat_b2a.py -q && .venv/bin/python -c "import yaml; yaml.safe_load_all(open('infra/k8s/jobs/fxhnt-ucits-rebalancer.yaml')); print('yaml OK')"`
Expected: PASS (default `--venue us` path unchanged) + `yaml OK`; confirm `suspend: true`.

- [ ] **Step 7: Commit**

```bash
git add src/fxhnt/config.py src/fxhnt/application/ucits_map.py src/fxhnt/cli.py infra/k8s/jobs/fxhnt-ucits-rebalancer.yaml tests/unit/test_ucits_routing.py
git commit -m "feat(ucits): US->UCITS execute routing (--venue ucits) + suspended monthly CronJob"
```

---

## Task 9: Local cockpit verification (both components)

**Files:**
- Test: `tests/integration/test_cockpit_vrp_render.py`

**Interfaces:**
- Consumes: `ForwardNavRepo`, `create_app` (existing), the `vrp` registry entry (Task 5).

- [ ] **Step 1: Write the failing test** (seed a `vrp` forward row + ref, assert the detail renders)

```python
# tests/integration/test_cockpit_vrp_render.py
import datetime as dt
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from fxhnt.adapters.persistence.cockpit_models import ForwardSummaryRow
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.adapters.web.app import create_app


def test_vrp_detail_renders(tmp_path):
    repo = ForwardNavRepo(f"sqlite:///{tmp_path/'c.db'}")
    repo.migrate()
    at = dt.datetime(2026, 7, 12)
    with Session(repo._engine) as s:
        s.add(ForwardSummaryRow(strategy_id="vrp", as_of="2026-07-12", days=8, nav=1.004,
                                total_return=0.004, sharpe=0.9, maxdd=-0.03, gate_status="WAIT",
                                gate_reason="building 8/21", src_updated_at=at))
        s.commit()
    r = TestClient(create_app(repo)).get("/strategy/vrp")
    assert r.status_code == 200
    assert "Equity VRP" in r.text
    assert "ibkr-xsp" in r.text or "XSP" in r.text
```

- [ ] **Step 2: Run to verify it fails, then passes** once the `vrp` registry entry is present (Task 5 already added it — this test confirms the cockpit renders the sleeve).

Run: `.venv/bin/python -m pytest tests/integration/test_cockpit_vrp_render.py -q`
Expected: PASS.

- [ ] **Step 3: Manual local browser verify** (per `feedback_verify_cockpit_locally_not_prod`): run `scripts/dev-cockpit-local.sh`, seed a `vrp` recompute into the local dev DB, open `/strategy/vrp` in a real browser, screenshot, and READ: NAV line present, gate shows "building N/21", the real backtest ref is shown. Confirm the multistrat detail's `ibkr-ucits` badge path renders when a `vrp_exec`/`multistrat_exec` venue is `ibkr-ucits`. Do NOT claim "works" off synthetic events.

- [ ] **Step 4: Commit**

```bash
git add tests/integration/test_cockpit_vrp_render.py
git commit -m "test(vrp): cockpit renders the vrp sleeve detail (local verify)"
```

---

## Self-Review

**Spec coverage:**
- A.1 data path → Task 2 (OPRA adapter) + Task 3 (PIT table) + Task 6 (freeze slice). ✓
- A.2 parity forward + IV inversion → Task 1 (helpers) + Task 6 (`_xsp_forward_estimate`). ✓
- A.3 strategy rules (weekly, 30 DTE, 20Δ, 5-pt width, ladder, 50%/expiry-1, defined-risk return) → Task 4. ✓
- A.4 forward/gate/sizing (vrp_nav, registry, migration anchor, CVaR test) → Tasks 4/5/6. ✓
- A.5 exec twin (IBKR mleg, execute-vrp, suspended CronJob, vrp_exec) → Task 7. ✓
- B.1 UCITS map (config) → Task 8. ✓
- B.2 execute path (`--venue ucits`, unmapped refusal, whole-share, suspended monthly CronJob) → Task 8. ✓
- Data flow / error handling / testing (defined-risk bound, degenerate IV skip, ref-row written, local cockpit) → Tasks 4/6/9. ✓
- One-time 2013→ backfill Job (no startup DDL) → Task 7 (`fxhnt-opra-backfill.yaml`). ✓

**Placeholder scan:** no "TBD/TODO/handle edge cases" — every code step has real code; Task 7 Step 5 and Task 3/8 CronJobs describe copy-from-existing-file diffs with the exact fields to change (name/schedule/suspend/command), which is concrete.

**Type consistency:** `OptionContract(osi_symbol, expiry, strike, right)` consistent across Tasks 2/3/4/6. `bars_for` → `{osi: {iso: close}}` consistent (Tasks 3/4/6). `fetch_option_bars` returns the same `{osi: {iso: close}}` shape (Task 2/6). `build_combo_legs` legs `(osi, side, qty)` match `place_combo` (Task 7). `apply_ucits_map(weights, prices, mapping)` consistent (Task 8). `VrpStrategy` ctor kwargs match the registry `definition.params` (Tasks 4/5). `margin = width_points*100` consistent (Global Constraints / Task 4).

**One flagged risk for the implementer (Task 4 `_select_spread`):** when the ATM call is absent from the frozen slice, `_select_spread` approximates the forward by the ATM put strike; the asset's `_xsp_forward_estimate` uses the true parity forward. If a review finds the strategy's internal forward approximation materially shifts the 20Δ selection vs the asset's parity forward, pass the parity forward into `VrpStrategy` via `extra` (freeze it alongside the slice) rather than re-deriving inside the strategy — surface this to the human if it arises rather than silently diverging the two forwards.
