# Lifecycle B0 — Dagster Foundation + Combined-Book via Warehouse — 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:** Replace the `fxhnt-forward` cron with a Dagster asset graph (on the cluster) that materializes the combined-book daily paper-forward track from the warehouse SSOT and ingests it to the cockpit.

**Architecture:** Refactor `CombinedBook` to read through an injectable `CombinedBookSource` port (`WarehouseSource` reads the SSOT, `RawFileSource` is the existing fallback). Four Dagster assets — `crypto_bars`, `futures_bars` → `combined_forward_nav` → `cockpit_forward` — wrap existing services; a daily Dagster schedule replaces the cron. Dagster OSS deploys on the cluster (webserver + daemon + storage in the existing cockpit Postgres), tailnet-private.

**Tech Stack:** Python 3.11, numpy, DuckDB (warehouse), Postgres (cockpit + Dagster storage), Dagster OSS, Kubernetes/Argo. Tests: `python3 -m pytest <path> -q` (pythonpath=`src`), `mypy strict = true`.

**Reference spec:** `docs/superpowers/specs/2026-06-15-lifecycle-dagster-b0-design.md`

**Dagster note:** Dagster is an external dependency the engineer may not know. For Tasks 7–10, pull the current Dagster API via context7 (`mcp__context7__resolve-library-id "dagster"` → `query-docs` for `@asset`, `Definitions`, `ScheduleDefinition`, `FreshnessPolicy`, Postgres storage, deployment) and verify against the installed version before finalizing — the code below targets Dagster 1.8+ and is standard, but confirm signatures.

**Convention:** epoch-day = `ts_seconds // 86400`. The warehouse stores `ts` in epoch-seconds; the crypto panel uses epoch-days (matches `load_crypto_panel`). Tests under `tests/unit/` and `tests/integration/`.

---

## Task 1: Factor the panel builder out of `load_crypto_panel`

**Files:**
- Modify: `src/fxhnt/application/crypto_momentum_research.py:23-40`
- Test: `tests/unit/test_crypto_panel_builder.py`

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

```python
# tests/unit/test_crypto_panel_builder.py
"""The panel builder turns {sym: {day: close}} into aligned (days, syms, close, ret) — shared by the
raw-file and warehouse sources so they produce identical panels."""
from __future__ import annotations

import numpy as np

from fxhnt.application.crypto_momentum_research import _panel


def test_panel_aligns_and_computes_returns() -> None:
    data = {"BTCUSDT": {19000: 100.0, 19001: 110.0}, "ETHUSDT": {19001: 50.0, 19002: 55.0}}
    days, syms, close, ret = _panel(data)
    assert days == [19000, 19001, 19002]
    assert syms == ["BTCUSDT", "ETHUSDT"]
    assert np.isnan(close[0, 1]) and close[1, 0] == 110.0 and close[2, 1] == 55.0
    assert abs(ret[1, 0] - 0.10) < 1e-12          # 110/100 - 1
    assert np.isnan(ret[0, 0])


def test_panel_empty() -> None:
    days, syms, close, ret = _panel({})
    assert days == [] and syms == [] and close.shape == (0, 0)
```

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

Run: `python3 -m pytest tests/unit/test_crypto_panel_builder.py -q`
Expected: FAIL — `cannot import name '_panel'`.

- [ ] **Step 3: Implement** — in `src/fxhnt/application/crypto_momentum_research.py`, extract the matrix/ret logic into `_panel` and make `load_crypto_panel` call it:

```python
def _panel(data: dict[str, dict[int, float]]) -> tuple[list[int], list[str], np.ndarray, np.ndarray]:
    """Aligned panel from {symbol: {epoch_day: close}} → (days, syms, close[d×s], ret[d×s])."""
    syms = sorted(data)
    days = sorted(set().union(*(set(d) for d in data.values()))) if data else []
    di = {d: i for i, d in enumerate(days)}
    close = np.full((len(days), len(syms)), np.nan)
    for j, s in enumerate(syms):
        for d, c in data[s].items():
            if c and c > 0:
                close[di[d], j] = c
    ret = np.full_like(close, np.nan)
    if len(days) > 1:
        ret[1:] = close[1:] / close[:-1] - 1.0
    return days, syms, close, ret


def load_crypto_panel(crypto_dir: str) -> tuple[list[int], list[str], np.ndarray, np.ndarray]:
    """Load a survivorship-free crypto close panel from a dir of .npz (keys: day, close)."""
    import glob
    import os

    data: dict[str, dict[int, float]] = {}
    for f in sorted(glob.glob(os.path.join(crypto_dir, "*.npz"))):
        z = np.load(f)   # numeric-only arrays (our own surfer-fetch files); no allow_pickle needed
        if "close" in z and "day" in z:
            data[os.path.basename(f)[:-4]] = dict(zip(z["day"].tolist(), z["close"].tolist(), strict=True))
    return _panel(data)
```

(Keep the existing module-level `import glob`/`import os` if present; the inline imports above are only if they were function-local. Match the file.)

- [ ] **Step 4: Run tests + full suite**

Run: `python3 -m pytest tests/unit/test_crypto_panel_builder.py -q` (PASS), then `python3 -m pytest tests/ -q` (no regressions — existing crypto-panel callers unchanged).

- [ ] **Step 5: mypy + commit**

Run: `python3 -m mypy src/fxhnt/application/crypto_momentum_research.py`
```bash
git add src/fxhnt/application/crypto_momentum_research.py tests/unit/test_crypto_panel_builder.py
git commit -m "refactor(crypto): extract _panel builder (shared by raw + warehouse sources)"
```

---

## Task 2: Warehouse `read_panel(symbols, feature)`

**Files:**
- Modify: `src/fxhnt/adapters/warehouse/duckdb_feature_store.py` (add `read_panel` before `close`)
- Test: `tests/integration/test_warehouse_read_panel.py`

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

```python
# tests/integration/test_warehouse_read_panel.py
"""read_panel returns per-symbol {epoch_day: value} dicts for one feature — the crypto-panel read path."""
from __future__ import annotations

from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
from fxhnt.application.warehouse_ingest import WarehouseIngest
from fxhnt.domain.warehouse.bar import BarRecord


def test_read_panel_groups_by_symbol_and_day(tmp_path) -> None:
    store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    WarehouseIngest(store).ingest_bars("BTCUSDT", [BarRecord("BTCUSDT", 19000 * 86400, 100.0, 100.0, 100.0),
                                                   BarRecord("BTCUSDT", 19001 * 86400, 110.0, 110.0, 110.0)])
    WarehouseIngest(store).ingest_bars("ETHUSDT", [BarRecord("ETHUSDT", 19001 * 86400, 50.0, 50.0, 50.0)])
    panel = store.read_panel(["BTCUSDT", "ETHUSDT"], "close")
    assert panel["BTCUSDT"] == {19000: 100.0, 19001: 110.0}
    assert panel["ETHUSDT"] == {19001: 50.0}
    store.close()
```

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

Run: `python3 -m pytest tests/integration/test_warehouse_read_panel.py -q`
Expected: FAIL — `DuckDbFeatureStore` has no `read_panel`.

- [ ] **Step 3: Implement** — add to `src/fxhnt/adapters/warehouse/duckdb_feature_store.py` (before `close`):

```python
    def read_panel(self, symbols: list[str], feature: str) -> dict[str, dict[int, float]]:
        """{symbol: {epoch_day: value}} for one feature across symbols — the cross-sectional panel read.
        ts (epoch-seconds) is converted to epoch-days (ts // 86400)."""
        if not symbols:
            return {}
        placeholders = ",".join("?" for _ in symbols)
        rows = self._con.execute(
            f"SELECT symbol, ts, value FROM features WHERE feature = ? AND symbol IN ({placeholders})",
            [feature, *symbols],
        ).fetchall()
        out: dict[str, dict[int, float]] = {s: {} for s in symbols}
        for symbol, ts, value in rows:
            out[symbol][int(ts) // 86_400] = float(value)
        return out
```

- [ ] **Step 4: Run test + mypy**

Run: `python3 -m pytest tests/integration/test_warehouse_read_panel.py -q` (PASS), `python3 -m mypy src/fxhnt/adapters/warehouse/duckdb_feature_store.py`.

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/adapters/warehouse/duckdb_feature_store.py tests/integration/test_warehouse_read_panel.py
git commit -m "feat(warehouse): read_panel — per-symbol epoch-day matrix for one feature"
```

---

## Task 3: `CombinedBookSource` port + `RawFileSource`

**Files:**
- Create: `src/fxhnt/application/combined_book_source.py`
- Test: `tests/unit/test_combined_book_source.py`

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

```python
# tests/unit/test_combined_book_source.py
"""CombinedBookSource: RawFileSource reproduces load_crypto_panel; WarehouseSource (Task 4) matches it."""
from __future__ import annotations

import numpy as np


def test_raw_file_source_crypto_panel_matches_loader(tmp_path) -> None:
    from fxhnt.application.combined_book_source import RawFileSource
    cp = tmp_path / "crypto_pit"; cp.mkdir()
    np.savez(cp / "BTCUSDT.npz", day=np.array([19000, 19001]), close=np.array([100.0, 110.0]))
    np.savez(cp / "ETHUSDT.npz", day=np.array([19001]), close=np.array([50.0]))
    days, syms, close, ret = RawFileSource(str(tmp_path)).crypto_panel()
    assert days == [19000, 19001] and syms == ["BTCUSDT", "ETHUSDT"]
    assert close[1, 0] == 110.0 and abs(ret[1, 0] - 0.10) < 1e-12
```

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

Run: `python3 -m pytest tests/unit/test_combined_book_source.py -q`
Expected: FAIL — `ModuleNotFoundError: fxhnt.application.combined_book_source`.

- [ ] **Step 3: Implement** — create `src/fxhnt/application/combined_book_source.py`:

```python
"""Data source for the combined book — the A→B seam. The book reads its crypto panel + futures series
through this port; RawFileSource is the existing raw-file path, WarehouseSource (Task 4) reads the SSOT."""
from __future__ import annotations

from typing import Protocol

import numpy as np

from fxhnt.adapters.data.dbn_local import LocalDbnLoader
from fxhnt.application.crypto_momentum_research import load_crypto_panel
from fxhnt.domain.models import PriceSeries


class CombinedBookSource(Protocol):
    def crypto_panel(self) -> tuple[list[int], list[str], np.ndarray, np.ndarray]:
        """(days, syms, close[d×s], ret[d×s]) — same shape as load_crypto_panel."""
        ...

    def futures_series(self, root: str) -> PriceSeries:
        """Continuous front-month PriceSeries for a futures root."""
        ...


class RawFileSource:
    def __init__(self, data_dir: str, crypto_subdir: str = "crypto_pit") -> None:
        self._dir = data_dir
        self._sub = crypto_subdir
        self._loader = LocalDbnLoader()

    def crypto_panel(self) -> tuple[list[int], list[str], np.ndarray, np.ndarray]:
        return load_crypto_panel(f"{self._dir}/{self._sub}")

    def futures_series(self, root: str) -> PriceSeries:
        return self._loader.load(f"{self._dir}/{root}.dbn", root)
```

- [ ] **Step 4: Run test + mypy**

Run: `python3 -m pytest tests/unit/test_combined_book_source.py -q` (PASS), `python3 -m mypy src/fxhnt/application/combined_book_source.py`.

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/combined_book_source.py tests/unit/test_combined_book_source.py
git commit -m "feat(combined-book): CombinedBookSource port + RawFileSource (existing path)"
```

---

## Task 4: `WarehouseSource` + the equivalence guarantee

**Files:**
- Modify: `src/fxhnt/application/combined_book_source.py`
- Test: `tests/integration/test_combined_book_source_equivalence.py`

- [ ] **Step 1: Write the failing test (THE A→B correctness gate)**

```python
# tests/integration/test_combined_book_source_equivalence.py
"""WarehouseSource.crypto_panel must equal RawFileSource.crypto_panel on the same data — proves the
SSOT swap is faithful (the A→B guarantee)."""
from __future__ import annotations

import numpy as np

from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
from fxhnt.application.combined_book_source import RawFileSource, WarehouseSource
from fxhnt.application.silver_ingest_helpers import ingest_crypto_pit  # Task helper below


def test_warehouse_panel_equals_rawfile_panel(tmp_path) -> None:
    cp = tmp_path / "crypto_pit"; cp.mkdir()
    np.savez(cp / "BTCUSDT.npz", day=np.array([19000, 19001, 19002]), close=np.array([100.0, 110.0, 105.0]))
    np.savez(cp / "ETHUSDT.npz", day=np.array([19001, 19002]), close=np.array([50.0, 55.0]))
    store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    ingest_crypto_pit(store, str(cp))                         # bronze .npz → silver → warehouse

    d_raw, s_raw, c_raw, r_raw = RawFileSource(str(tmp_path)).crypto_panel()
    d_wh, s_wh, c_wh, r_wh = WarehouseSource(store).crypto_panel()
    assert d_raw == d_wh and s_raw == s_wh
    assert np.allclose(c_raw, c_wh, equal_nan=True)
    assert np.allclose(r_raw, r_wh, equal_nan=True)
    store.close()
```

- [ ] **Step 2: Add the ingest helper used by the test and the assets**

Create `src/fxhnt/application/silver_ingest_helpers.py` (small, reused by Task 8 assets):

```python
"""Bronze→silver→warehouse ingest helpers for the orchestration assets + tests (one place so the asset
and the equivalence test ingest identically)."""
from __future__ import annotations

import glob
import os

import numpy as np

from fxhnt.adapters.warehouse.silver_crypto import normalize_crypto
from fxhnt.application.warehouse_ingest import WarehouseIngest
from fxhnt.ports.warehouse import FeatureStore


def ingest_crypto_pit(store: FeatureStore, crypto_dir: str) -> int:
    """Ingest every crypto_pit/<SYM>.npz (keys day, close) into the warehouse via silver_crypto."""
    ingest = WarehouseIngest(store)
    total = 0
    for path in sorted(glob.glob(os.path.join(crypto_dir, "*.npz"))):
        sym = os.path.splitext(os.path.basename(path))[0]
        with np.load(path) as z:
            total += ingest.ingest_bars(sym, normalize_crypto(sym, z["day"], z["close"]))
    return total
```

- [ ] **Step 3: Run test to verify it fails**

Run: `python3 -m pytest tests/integration/test_combined_book_source_equivalence.py -q`
Expected: FAIL — `cannot import name 'WarehouseSource'`.

- [ ] **Step 4: Implement `WarehouseSource`** — append to `src/fxhnt/application/combined_book_source.py` (add imports at top):

```python
from fxhnt.adapters.warehouse.price_provider import WarehousePriceProvider
from fxhnt.application.crypto_momentum_research import _panel
from fxhnt.domain.models import AssetClass, Market
from fxhnt.ports.warehouse import FeatureStore


class WarehouseSource:
    """Reads the combined book's data from the warehouse SSOT. Crypto = the panel of all USDT-perp
    symbols in the warehouse; futures = per-root continuous front via WarehousePriceProvider."""

    def __init__(self, store: FeatureStore) -> None:
        self._store = store
        self._provider = WarehousePriceProvider(store)

    def _crypto_symbols(self) -> list[str]:
        return [s for s, *_ in self._store.catalog() if s.endswith("USDT")]

    def crypto_panel(self) -> tuple[list[int], list[str], np.ndarray, np.ndarray]:
        data = self._store.read_panel(self._crypto_symbols(), "close")
        return _panel(data)

    def futures_series(self, root: str) -> PriceSeries:
        return self._provider.fetch(Market(symbol=root, asset_class=AssetClass.FUTURE))
```

Note: `WarehouseSource` requires `FeatureStore` to expose `catalog()` (Task DL-8, present) and `read_panel` (Task 2). The equivalence holds because both sources funnel through the shared `_panel`, and `silver_crypto` writes `close` unchanged (so warehouse close == .npz close).

- [ ] **Step 5: Run test + mypy**

Run: `python3 -m pytest tests/integration/test_combined_book_source_equivalence.py -q` (PASS — the A→B guarantee), `python3 -m mypy src/fxhnt/application/combined_book_source.py src/fxhnt/application/silver_ingest_helpers.py`.

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/application/combined_book_source.py src/fxhnt/application/silver_ingest_helpers.py \
        tests/integration/test_combined_book_source_equivalence.py
git commit -m "feat(combined-book): WarehouseSource reads SSOT; equivalence test proves A->B faithful"
```

---

## Task 5: Inject the source into `CombinedBook`

**Files:**
- Modify: `src/fxhnt/application/combined_book.py:47-72`
- Test: `tests/unit/test_combined_book_injected_source.py`

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

```python
# tests/unit/test_combined_book_injected_source.py
"""CombinedBook reads through an injected CombinedBookSource; default keeps the raw-file behavior."""
from __future__ import annotations

import numpy as np

from fxhnt.application.combined_book import CombinedBook
from fxhnt.domain.models import AssetClass, Market, PriceSeries


class _FakeSource:
    def crypto_panel(self):
        days = list(range(19000, 19000 + 400))
        close = np.cumprod(1.0 + np.full((400, 3), 0.001), axis=0) * 100.0
        ret = np.full_like(close, np.nan); ret[1:] = close[1:] / close[:-1] - 1.0
        return days, ["AUSDT", "BUSDT", "CUSDT"], close, ret

    def futures_series(self, root):
        dates = tuple(f"2020-{(i % 12) + 1:02d}-{(i % 27) + 1:02d}" for i in range(400))
        close = np.cumprod(1.0 + np.full(400, 0.0005)) * 100.0
        return PriceSeries(market=Market(symbol=root, asset_class=AssetClass.FUTURE), dates=dates, close=close)


def test_combined_book_uses_injected_source() -> None:
    book = CombinedBook("unused", ["ES"], source=_FakeSource())
    res, dates, combo = book.build()
    assert res.overlap_days > 0 and len(combo) == len(dates)


def test_default_source_is_rawfile(tmp_path) -> None:
    # constructing without a source must not raise (defaults to RawFileSource(data_dir))
    book = CombinedBook(str(tmp_path), ["ES"])
    assert book is not None
```

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

Run: `python3 -m pytest tests/unit/test_combined_book_injected_source.py -q`
Expected: FAIL — `CombinedBook.__init__` has no `source` kwarg.

- [ ] **Step 3: Implement** — edit `src/fxhnt/application/combined_book.py`. Replace the import of `LocalDbnLoader`/`load_crypto_panel` usage and the `__init__`/`_crypto`/`_trend` (lines 47-72) with:

```python
from fxhnt.application.combined_book_source import CombinedBookSource, RawFileSource
# (keep: from fxhnt.application.crypto_momentum_research import xs_momentum_book
#        from fxhnt.application.tsmom_research import _book_returns, vol_targeted_trend)
# (REMOVE the now-unused: LocalDbnLoader import, load_crypto_panel import)


class CombinedBook:
    def __init__(self, data_dir: str, futures: list[str], *, source: CombinedBookSource | None = None,
                 crypto_subdir: str = "crypto_pit", crypto_lookback: int = 30, crypto_cost_bps: float = 8.0,
                 futures_cost_bps: float = 1.0, target_vol: float = 0.15) -> None:
        self._futures = futures
        self._source = source if source is not None else RawFileSource(data_dir, crypto_subdir)
        self._cl, self._cc, self._fc, self._tv = crypto_lookback, crypto_cost_bps, futures_cost_bps, target_vol

    def _crypto(self) -> dict[str, float]:
        days, _, close, ret = self._source.crypto_panel()
        book = xs_momentum_book(close, ret, self._cl, cost_bps=self._cc)
        keys = [dt.datetime.fromtimestamp(int(d) * 86400, dt.timezone.utc).strftime("%Y-%m-%d") for d in days]
        return dict(zip(keys, book))

    def _trend(self) -> dict[str, float]:
        sleeves = []
        for s in self._futures:
            try:
                ps = self._source.futures_series(s)
            except Exception:
                continue
            sleeves.append(vol_targeted_trend(ps, (20, 60, 120, 250), cost_bps=self._fc))
        dates, book = _book_returns(sleeves)
        return dict(zip(dates, book))
```

Leave `build()` unchanged. (`dt` is already imported in the file.)

- [ ] **Step 4: Run test + full suite + mypy**

Run: `python3 -m pytest tests/unit/test_combined_book_injected_source.py -q` (PASS), `python3 -m pytest tests/ -q` (no regressions — the `forward-track` CLI still constructs `CombinedBook(data_dir, futures)` → defaults to RawFileSource), `python3 -m mypy src/fxhnt/application/combined_book.py`.

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/combined_book.py tests/unit/test_combined_book_injected_source.py
git commit -m "refactor(combined-book): read via injected CombinedBookSource (default RawFileSource)"
```

---

## Task 6: Config flag + source factory

**Files:**
- Modify: `src/fxhnt/config.py` (add field); Create: `src/fxhnt/application/combined_book_factory.py`
- Test: `tests/unit/test_combined_book_factory.py`

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

```python
# tests/unit/test_combined_book_factory.py
"""The factory picks WarehouseSource vs RawFileSource from settings (FXHNT_COMBINED_BOOK_DATA_SOURCE)."""
from __future__ import annotations

from fxhnt.application.combined_book_factory import make_combined_book_source
from fxhnt.application.combined_book_source import RawFileSource, WarehouseSource
from fxhnt.config import Settings


def test_factory_raw_default(tmp_path) -> None:
    s = Settings(combined_book_data_source="raw", warehouse_path=str(tmp_path / "wh.duckdb"))
    assert isinstance(make_combined_book_source(s, "/data/surfer"), RawFileSource)


def test_factory_warehouse(tmp_path) -> None:
    s = Settings(combined_book_data_source="warehouse", warehouse_path=str(tmp_path / "wh.duckdb"))
    assert isinstance(make_combined_book_source(s, "/data/surfer"), WarehouseSource)
```

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

Run: `python3 -m pytest tests/unit/test_combined_book_factory.py -q`
Expected: FAIL — no `combined_book_data_source` / no factory module.

- [ ] **Step 3: Implement** — in `src/fxhnt/config.py` add the field to `Settings` (after `warehouse_path`):

```python
    # Combined-book data path: "raw" (crypto_pit/.dbn files) or "warehouse" (SSOT). Default raw for safety;
    # the B0 Dagster deploy sets FXHNT_COMBINED_BOOK_DATA_SOURCE=warehouse.
    combined_book_data_source: str = Field(default="raw")
```

Create `src/fxhnt/application/combined_book_factory.py`:

```python
"""Composition helper: build the CombinedBookSource the settings select (the reversible cutover seam)."""
from __future__ import annotations

from fxhnt.application.combined_book_source import CombinedBookSource, RawFileSource, WarehouseSource
from fxhnt.config import Settings


def make_combined_book_source(settings: Settings, data_dir: str) -> CombinedBookSource:
    if settings.combined_book_data_source == "warehouse":
        from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
        return WarehouseSource(DuckDbFeatureStore(settings.warehouse_path))
    return RawFileSource(data_dir)
```

- [ ] **Step 4: Run test + mypy**

Run: `python3 -m pytest tests/unit/test_combined_book_factory.py -q` (PASS), `python3 -m mypy src/fxhnt/config.py src/fxhnt/application/combined_book_factory.py`.

- [ ] **Step 5: Wire the `forward-track` CLI to honor the flag**

In `src/fxhnt/cli.py` `forward_track`, replace `book = CombinedBook(data_dir, _FUTURES_UNIVERSE)` with:

```python
        from fxhnt.application.combined_book_factory import make_combined_book_source
        book = CombinedBook(data_dir, _FUTURES_UNIVERSE,
                            source=make_combined_book_source(get_settings(), data_dir))
```

Run: `python3 -m pytest tests/ -q` (no regressions), `python3 -m mypy src/fxhnt/cli.py` (no NEW errors vs the known pre-existing ones).

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/config.py src/fxhnt/application/combined_book_factory.py src/fxhnt/cli.py \
        tests/unit/test_combined_book_factory.py
git commit -m "feat(combined-book): FXHNT_COMBINED_BOOK_DATA_SOURCE flag + source factory; wire CLI"
```

---

## Task 7: Add Dagster dependency

**Files:**
- Modify: `pyproject.toml` (add `[orchestration]` extra)
- Test: import check

- [ ] **Step 1: Add the extra** — in `pyproject.toml` `[project.optional-dependencies]`:

```toml
orchestration = ["dagster>=1.8,<1.9", "dagster-webserver>=1.8,<1.9", "dagster-postgres>=0.24,<0.25"]
```

- [ ] **Step 2: Install + verify**

Run: `pip install -e '.[orchestration]'` then `python3 -c "import dagster; print(dagster.__version__)"`
Expected: prints a 1.8.x version. (If the resolver picks a different compatible line, note it and pin accordingly — then re-confirm the `@asset`/`Definitions`/`ScheduleDefinition` API via context7 for that exact version.)

- [ ] **Step 3: Commit**

```bash
git add pyproject.toml
git commit -m "build(orchestration): add dagster optional-dependency extra"
```

---

## Task 8: Dagster assets

**Files:**
- Create: `src/fxhnt/adapters/orchestration/__init__.py`, `src/fxhnt/adapters/orchestration/assets.py`
- Test: `tests/integration/test_orchestration_assets.py`

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

```python
# tests/integration/test_orchestration_assets.py
"""The combined_forward_nav asset materializes from the warehouse + books into a state file; cockpit
ingest is idempotent. Assets are invoked directly (no live Dagster instance)."""
from __future__ import annotations

import numpy as np

from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
from fxhnt.application.combined_book_source import WarehouseSource
from fxhnt.application.silver_ingest_helpers import ingest_crypto_pit
from fxhnt.adapters.orchestration.assets import build_combined_forward_nav


def test_combined_forward_nav_materializes_from_warehouse(tmp_path) -> None:
    cp = tmp_path / "crypto_pit"; cp.mkdir()
    rng = np.random.default_rng(0)
    for sym in ("BTCUSDT", "ETHUSDT", "SOLUSDT"):
        days = np.arange(19000, 19000 + 400, dtype=np.int64)
        close = np.cumprod(1.0 + rng.normal(0.0005, 0.02, 400)) * 100.0
        np.savez(cp / f"{sym}.npz", day=days, close=close)
    store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    ingest_crypto_pit(store, str(cp))
    # build_combined_forward_nav is the plain function behind the @asset (no futures → crypto-only book ok)
    summary = build_combined_forward_nav(WarehouseSource(store), futures=[], state_path=str(tmp_path / "fwd.json"))
    assert summary.forward_days >= 0 and summary.inception is not None
    store.close()
```

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

Run: `python3 -m pytest tests/integration/test_orchestration_assets.py -q`
Expected: FAIL — `ModuleNotFoundError: fxhnt.adapters.orchestration.assets`.

- [ ] **Step 3: Implement** — create `src/fxhnt/adapters/orchestration/__init__.py` (empty docstring) and `assets.py`. Keep plain functions behind each `@asset` so they are unit-testable without Dagster:

```python
# src/fxhnt/adapters/orchestration/__init__.py
"""Dagster orchestration adapter — the lifecycle asset graph (B0: the combined-book track)."""
```

```python
# src/fxhnt/adapters/orchestration/assets.py
"""B0 asset graph: crypto_bars + futures_bars → combined_forward_nav → cockpit_forward.
Each @asset is a thin wrapper over a plain function (also unit-testable without a Dagster instance)."""
from __future__ import annotations

from dagster import AssetExecutionContext, Config, FreshnessPolicy, asset

from fxhnt.application.combined_book import CombinedBook
from fxhnt.application.combined_book_source import CombinedBookSource
from fxhnt.application.forward_tracker import CombinedBookForwardTracker, ForwardStatus

_FUTURES = ["ES", "NQ", "YM", "RTY", "EMD", "NKD"]


def build_combined_forward_nav(source: CombinedBookSource, futures: list[str], state_path: str) -> ForwardStatus:
    """Plain function behind the combined_forward_nav asset — book today's combined-book return."""
    book = CombinedBook("warehouse", futures, source=source)
    return CombinedBookForwardTracker(book, state_path).step()


# ---- Dagster assets (thin wrappers; resources injected via Definitions in Task 9) ----

@asset(freshness_policy=FreshnessPolicy(maximum_lag_minutes=60 * 36))
def crypto_bars(context: AssetExecutionContext) -> int:
    from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
    from fxhnt.application.silver_ingest_helpers import ingest_crypto_pit
    from fxhnt.config import get_settings
    s = get_settings(); s.ensure_dirs()
    n = ingest_crypto_pit(DuckDbFeatureStore(s.warehouse_path), f"{_data_dir()}/crypto_pit")
    context.log.info(f"crypto_bars ingested {n} rows")
    return n


@asset(freshness_policy=FreshnessPolicy(maximum_lag_minutes=60 * 24 * 8))
def futures_bars(context: AssetExecutionContext) -> int:
    from fxhnt.adapters.data.dbn_local import read_front_arrays
    from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
    from fxhnt.adapters.warehouse.silver_futures import normalize_futures
    from fxhnt.application.warehouse_ingest import WarehouseIngest
    from fxhnt.config import get_settings
    s = get_settings(); s.ensure_dirs()
    ingest = WarehouseIngest(DuckDbFeatureStore(s.warehouse_path))
    total = 0
    for root in _FUTURES:
        try:
            days, inst, close, high, low = read_front_arrays(f"{_data_dir()}/{root}.dbn", root)
            total += ingest.ingest_bars(root, normalize_futures(root, days, inst, close, high, low))
        except Exception as e:                       # a missing root must not fail the whole asset
            context.log.warning(f"futures_bars {root} skipped: {e}")
    context.log.info(f"futures_bars ingested {total} rows")
    return total


@asset(deps=[crypto_bars, futures_bars])
def combined_forward_nav(context: AssetExecutionContext) -> dict:
    from fxhnt.application.combined_book_factory import make_combined_book_source
    from fxhnt.config import get_settings
    s = get_settings()
    source = make_combined_book_source(s, _data_dir())
    st = build_combined_forward_nav(source, _FUTURES, f"{_data_dir()}/fxhnt_combined_forward.json")
    context.log.info(f"combined_forward_nav: {st.forward_days} forward days, booked through {st.last_date}")
    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}


@asset(deps=[combined_forward_nav])
def cockpit_forward(context: AssetExecutionContext) -> None:
    """Ingest the combined-book forward NAV into the cockpit Postgres (reuses the existing ingest path)."""
    from fxhnt.application.forward_ingest import ingest_forward_state   # existing ingest-forward logic
    from fxhnt.config import get_settings
    s = get_settings()
    ingest_forward_state(s.operational_dsn, f"{_data_dir()}")
    context.log.info("cockpit_forward ingested")


def _data_dir() -> str:
    import os
    return os.environ.get("FXHNT_SURFER_DATA_DIR", "/data/surfer")
```

NOTE for the implementer: `cockpit_forward` must call the SAME logic the existing `fxhnt ingest-forward` CLI uses. Read `cli.py`'s `ingest_forward` command and factor its body into `application/forward_ingest.py::ingest_forward_state(dsn, data_dir)` (a small extract-to-function refactor), then have both the CLI and this asset call it (DRY). If that function already exists under another name, use it. Verify the exact name before finalizing.

- [ ] **Step 4: Run test + mypy**

Run: `python3 -m pytest tests/integration/test_orchestration_assets.py -q` (PASS), `python3 -m mypy src/fxhnt/adapters/orchestration/assets.py`.

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/adapters/orchestration/ tests/integration/test_orchestration_assets.py \
        src/fxhnt/application/forward_ingest.py src/fxhnt/cli.py
git commit -m "feat(orchestration): B0 Dagster assets (crypto/futures bars -> nav -> cockpit)"
```

---

## Task 9: Dagster `Definitions` + daily schedule

**Files:**
- Create: `src/fxhnt/adapters/orchestration/definitions.py`
- Test: `tests/integration/test_orchestration_definitions.py`

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

```python
# tests/integration/test_orchestration_definitions.py
"""The Definitions load and expose a daily schedule over the combined-book asset job."""
from __future__ import annotations


def test_definitions_load_with_daily_schedule() -> None:
    from fxhnt.adapters.orchestration.definitions import defs
    schedules = list(defs.get_all_schedules()) if hasattr(defs, "get_all_schedules") else defs.schedules
    names = [s.name for s in schedules]
    assert any("combined" in n for n in names)
    asset_keys = {k.to_user_string() for k in defs.get_asset_graph().all_asset_keys} \
        if hasattr(defs, "get_asset_graph") else set()
    assert {"crypto_bars", "futures_bars", "combined_forward_nav", "cockpit_forward"} <= asset_keys
```

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

Run: `python3 -m pytest tests/integration/test_orchestration_definitions.py -q`
Expected: FAIL — no `definitions` module.

- [ ] **Step 3: Implement** — create `src/fxhnt/adapters/orchestration/definitions.py` (verify the exact `define_asset_job`/`ScheduleDefinition` API for the installed Dagster version via context7):

```python
"""Dagster Definitions for the B0 lifecycle graph — assets + a daily schedule that materializes the
combined-book track (replaces the fxhnt-forward cron)."""
from __future__ import annotations

from dagster import Definitions, ScheduleDefinition, define_asset_job

from fxhnt.adapters.orchestration.assets import (
    cockpit_forward,
    combined_forward_nav,
    crypto_bars,
    futures_bars,
)

combined_book_job = define_asset_job(
    name="combined_book_forward_job",
    selection=[crypto_bars, futures_bars, combined_forward_nav, cockpit_forward],
)

daily_combined_book_schedule = ScheduleDefinition(
    name="combined_book_daily",
    job=combined_book_job,
    cron_schedule="30 23 * * *",     # same slot the fxhnt-forward cron used (23:30 UTC)
    execution_timezone="UTC",
)

defs = Definitions(
    assets=[crypto_bars, futures_bars, combined_forward_nav, cockpit_forward],
    jobs=[combined_book_job],
    schedules=[daily_combined_book_schedule],
)
```

- [ ] **Step 4: Run test + mypy**

Run: `python3 -m pytest tests/integration/test_orchestration_definitions.py -q` (PASS — adjust the assertion to the installed Dagster's introspection API if needed), then `python3 -c "from fxhnt.adapters.orchestration.definitions import defs; print('defs ok')"` and `dagster definitions validate -m fxhnt.adapters.orchestration.definitions` (if the CLI is available).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/adapters/orchestration/definitions.py tests/integration/test_orchestration_definitions.py
git commit -m "feat(orchestration): Definitions + daily combined-book schedule"
```

---

## Task 10: Cluster deployment (Dagster on Kapsule) + suspend the cron

**Files:**
- Modify: `infra/docker/fxhnt.Dockerfile` (install `.[orchestration]`)
- Create: `infra/k8s/orchestration/dagster.yaml` (ConfigMap), `dagster-webserver.yaml`, `dagster-daemon.yaml`, `netpol.yaml`
- Modify: `infra/k8s/jobs/fxhnt-forward-cronjob.yaml` (suspend)
- Create: `infra/argo/orchestration-build-deploy.yaml` (or extend the cockpit build)

This task is infrastructure — verified by `kubectl`/Argo, not pytest. Each step is deterministic.

- [ ] **Step 1: Dockerfile** — in `infra/docker/fxhnt.Dockerfile`, change the install to include the extra (match the existing `pip install` line; it currently installs `.[web,pg,data,factory]`):

```dockerfile
RUN pip install --no-cache-dir '.[web,pg,data,factory,orchestration]'
```

- [ ] **Step 2: Dagster instance config** — create `infra/k8s/orchestration/dagster.yaml` (a ConfigMap holding `dagster.yaml`) pointing storage at the existing cockpit Postgres, database `dagster`:

```yaml
apiVersion: v1
kind: ConfigMap
metadata: {name: dagster-instance, namespace: foxhunt}
data:
  dagster.yaml: |
    storage:
      postgres:
        postgres_db:
          username: foxhunt
          password:
            env: PGPASSWORD
          hostname: postgres.foxhunt.svc.cluster.local
          db_name: dagster
          port: 5432
    run_coordinator:
      module: dagster.core.run_coordinator
      class: QueuedRunCoordinator
```

First create the `dagster` database once: `kubectl exec -n foxhunt deploy/fxhnt-dashboard -- sh -c 'PGPASSWORD=$PGPASSWORD createdb -h postgres.foxhunt.svc.cluster.local -U foxhunt dagster'` (ignore "already exists").

- [ ] **Step 3: Webserver + daemon Deployments** — create `infra/k8s/orchestration/dagster-webserver.yaml` and `dagster-daemon.yaml`. Both: image = the `fxhnt-cockpit` image; env `DAGSTER_HOME=/dagster-home` (mount the ConfigMap's `dagster.yaml` there), `PGPASSWORD` from `db-credentials`, `FXHNT_COMBINED_BOOK_DATA_SOURCE=warehouse`, `FXHNT_OPERATIONAL_DSN` (cockpit DSN), `FXHNT_WAREHOUSE_PATH`, `FXHNT_SURFER_DATA_DIR=/data/surfer`, and the surfer-data PVC mounted at `/data/surfer` (same PVC the fxhnt-forward cron used). Commands:
  - webserver: `dagster-webserver -h 0.0.0.0 -p 3000 -m fxhnt.adapters.orchestration.definitions`
  - daemon: `dagster-daemon run -m fxhnt.adapters.orchestration.definitions`
  Add a **tailscale userspace sidecar** to the webserver pod (copy the pattern from `infra/k8s/services/fxhnt-dashboard.yaml`) so the UI is tailnet-private; no public Service/Ingress.

- [ ] **Step 4: NetworkPolicy** — create `infra/k8s/orchestration/netpol.yaml` allowing the dagster pods egress to `postgres.foxhunt.svc` (and DNS), mirroring the cockpit netpol (podSelector-based, per the Cilium lesson — not ipBlock).

- [ ] **Step 5: Build + deploy via Argo**

Run the existing build-deploy pattern for the cockpit image (it now includes orchestration), e.g. `./scripts/argo-... ` (mirror `infra/argo/cockpit-build-deploy.yaml`). Then:
```bash
kubectl apply -f infra/k8s/orchestration/
kubectl rollout status deploy/dagster-webserver -n foxhunt
kubectl rollout status deploy/dagster-daemon -n foxhunt
```
Expected: both deployments Ready (webserver 2/2 with the tailscale sidecar, daemon 1/1).

- [ ] **Step 6: Verify the graph + one materialization**

- UI reachable via tailnet (the webserver's tailscale node), shows the 4 assets + the `combined_book_daily` schedule.
- Trigger one run: `kubectl exec -n foxhunt deploy/dagster-daemon -- dagster job execute -m fxhnt.adapters.orchestration.definitions -j combined_book_forward_job` (or via the UI "Materialize all").
- Confirm it lands in the cockpit: `kubectl exec -n foxhunt deploy/fxhnt-dashboard -- python3 -c "from fxhnt.cli import app"` is not enough — instead verify the cockpit page renders the combined-book forward row updated (the existing dashboard.fxhnt.ai page), or query the `forward_nav` table for a fresh `src_updated_at`.

- [ ] **Step 7: Suspend the old cron (reversible cutover)**

In `infra/k8s/jobs/fxhnt-forward-cronjob.yaml` set `spec.suspend: true`, then `kubectl apply -f infra/k8s/jobs/fxhnt-forward-cronjob.yaml`. Leave it suspended (not deleted) until ≥3 Dagster daily runs match the prior track.

- [ ] **Step 8: Commit**

```bash
git add infra/docker/fxhnt.Dockerfile infra/k8s/orchestration/ infra/k8s/jobs/fxhnt-forward-cronjob.yaml infra/argo/
git commit -m "deploy(orchestration): Dagster on Kapsule (webserver+daemon, tailnet, pg storage); suspend fxhnt-forward cron"
```

---

## Final verification

- [ ] **Full suite + mypy on changed code**

Run:
```bash
python3 -m pytest tests/ -q
python3 -m mypy src/fxhnt/application/combined_book_source.py src/fxhnt/application/combined_book.py \
  src/fxhnt/application/combined_book_factory.py src/fxhnt/adapters/orchestration \
  src/fxhnt/adapters/warehouse/duckdb_feature_store.py
```
Expected: all tests pass; mypy clean on the new/changed files.

- [ ] **A→B guarantee holds**

`tests/integration/test_combined_book_source_equivalence.py` green = warehouse-fed and raw-fed panels are identical → the SSOT swap is faithful. With `FXHNT_COMBINED_BOOK_DATA_SOURCE=warehouse`, the deployed Dagster graph runs the combined book off the warehouse, end-to-end into the cockpit.

- [ ] **Cutover safety**

The `fxhnt-forward` cron is suspended (not deleted); flipping `FXHNT_COMBINED_BOOK_DATA_SOURCE=raw` (env on the dagster deployments) reverts the data path without code change.
