# Lifecycle Pipeline B0 — Dagster Foundation + Combined-Book via Warehouse — Design Spec

**Date:** 2026-06-15
**Status:** Approved (design)
**Sub-project:** B (strategy-lifecycle orchestration), increment **B0** of B0–B4. Target architecture: the
edge-factory lifecycle (propose → backtest → gauntlet → paper-forward → promote → live) as a Dagster
asset graph. B0 is the foundation: stand up Dagster on the cluster and migrate ONE track
(the combined book) end-to-end as an asset graph, reading from the warehouse SSOT (sub-project A).

## 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 —
proving the A→B spine (warehouse → strategy → cockpit, orchestrated, with lineage + freshness) and
de-risking the Dagster choice itself before migrating the other tracks (B1+).

## Context — what exists today

- **The track:** `CombinedBook(data_dir, futures_universe)` (`application/combined_book.py`) reads a
  crypto panel via `crypto_momentum_research.load_crypto_panel(data_dir/crypto_pit)` and 6 futures roots
  (`ES, NQ, YM, RTY, EMD, NKD`) via `LocalDbnLoader.load(<root>.dbn)`; `.build()` → weights + result.
  `CombinedBookForwardTracker(book, state).step()` (`application/forward_tracker.py`) books the day's
  realized return into a JSON state. `fxhnt forward-track` runs the booking; `fxhnt ingest-forward`
  upserts into the cockpit Postgres (`ForwardNavRepo`, idempotent, TimescaleDB hypertable).
- **Orchestration today:** the `fxhnt-forward` K8s CronJob (`sh -c: fxhnt fetch → fxhnt forward-track →
  fxhnt ingest-forward`), daily 23:30. The cockpit (`fxhnt-dashboard`) reads the Postgres forward tables.
- **Warehouse SSOT (sub-project A):** `DuckDbFeatureStore` (`features(symbol, ts, feature, value)`,
  point-in-time), `WarehousePriceProvider` (DataProvider port → single-market `PriceSeries`),
  silver normalizers (`silver_crypto`, `silver_futures`), `WarehouseIngest`, `warehouse_path` config.
- **Dagster:** NOT installed — a new dependency + cluster deployment.

## Architecture — the B0 asset graph

```
crypto_bars   ─┐   bronze crypto_pit fetch → silver_crypto → WarehouseIngest (all coins)
futures_bars  ─┤   bronze .dbn per root → silver_futures → WarehouseIngest (ES,NQ,YM,RTY,EMD,NKD)
               ▼
combined_forward_nav   reads warehouse panel (WarehousePanelSource) → CombinedBook.build +
               │       CombinedBookForwardTracker.step → forward-NAV state
               ▼
cockpit_forward        ForwardNavRepo upsert → cockpit Postgres (cockpit reads it)
```

A single daily **schedule** (Dagster `dagster-daemon`) materializes the graph, replacing the
`fxhnt-forward` CronJob. Each asset carries a **freshness policy** (crypto_bars < ~1 day; futures_bars
< ~8 days — futures fetch is weekly for cost). Lineage `bars → nav → cockpit` is visible in the UI.

### Component 1 — warehouse-backed data source (the A→B seam)

`CombinedBook` currently hardcodes `load_crypto_panel` + `LocalDbnLoader`. Refactor it to depend on an
injected **data source port**:

`application/combined_book_source.py` (port + impls):
```python
class CombinedBookSource(Protocol):
    def crypto_panel(self) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]: ...   # days,_,close,ret
    def futures_series(self, root: str) -> PriceSeries: ...

class RawFileSource:        # existing behavior, kept for fallback/reversibility
    # wraps load_crypto_panel(dir/crypto_pit) + LocalDbnLoader().load(dir/<root>.dbn)

class WarehouseSource:      # the SSOT path (default in B0)
    # crypto_panel: read all crypto symbols' close from the warehouse → aligned (days, close, ret) matrix
    # futures_series: WarehousePriceProvider(store).fetch(Market(root, FUTURE))
```

`CombinedBook.__init__` takes a `CombinedBookSource` (default constructed from settings:
`WarehouseSource` when `FXHNT_DATA_SOURCE=warehouse`, else `RawFileSource`). `_crypto()`/`_trend()` call
the injected source instead of importing the loaders directly. This is the one substantive refactor; it
is what makes the strategy read the SSOT.

The crypto panel read needs a **warehouse panel reader** — read `close` features for a set of symbols,
align on the common `ts` grid → `(days, close_matrix, ret_matrix)`. Add `read_panel(symbols, feature)`
to the warehouse adapter (or a helper in `WarehouseSource`) built on the existing `read_features`.

### Component 2 — the Dagster assets

`adapters/orchestration/assets.py` (Dagster `@asset` definitions; thin wrappers over existing services):
- `crypto_bars` — calls the existing crypto bronze fetch + `silver_crypto.normalize_crypto` +
  `WarehouseIngest.ingest_bars` for each coin. FreshnessPolicy ~24h.
- `futures_bars` — per root: bronze `.dbn` (existing `fetch`) + `read_front_arrays` +
  `silver_futures.normalize_futures` + ingest. FreshnessPolicy ~8d.
- `combined_forward_nav` — depends on `crypto_bars`, `futures_bars`; constructs `CombinedBook(...,
  source=WarehouseSource(store))`, runs `CombinedBookForwardTracker(book, state).step()`, returns the
  forward summary.
- `cockpit_forward` — depends on `combined_forward_nav`; upserts via `ForwardNavRepo` into cockpit PG.

`adapters/orchestration/definitions.py` — the Dagster `Definitions` (assets + a daily
`ScheduleDefinition` + resources: warehouse path, cockpit DSN). Resources injected, not hardcoded.

### Component 3 — deployment (cluster, reuse)

Dagster OSS, three pieces on the cluster (`infra/k8s/`):
- `dagster-webserver` Deployment (UI) + Service, **tailnet-private** via a tailscale userspace sidecar
  (same pattern as `fxhnt-dashboard`); no public exposure.
- `dagster-daemon` Deployment (runs schedules/sensors).
- **Run/event/schedule storage = the existing cockpit Postgres**, a separate database `dagster`
  (Dagster's `DagsterPostgresStorage`), configured via `dagster.yaml` reading `PGHOST`/credentials from
  the existing `db-credentials` secret.
- **Code location** = the existing `fxhnt-cockpit` image, with `dagster`, `dagster-webserver`,
  `dagster-postgres` added to `pyproject.toml` `[orchestration]` extra and the Dockerfile. Build via the
  existing Argo/Kaniko `cockpit-build-deploy` pattern.
- Both deployments run `fxhnt`'s `Definitions` from the image.

### Component 4 — cutover & reversibility

- The `fxhnt-forward` CronJob is **suspended** (`suspend: true`), not deleted, until the Dagster graph
  has produced ≥3 consistent daily NAV bookings matching the prior track.
- `FXHNT_DATA_SOURCE=warehouse|raw` selects `WarehouseSource` vs `RawFileSource` in `combined_forward_nav`
  — if the warehouse-fed NAV diverges from the raw-fed NAV, flip to `raw` without code change.
- Dagster's daily schedule and the suspended cron never run the booking concurrently.

## Data flow + idempotency

```
schedule (daily) → crypto_bars + futures_bars (bronze fetch → silver → warehouse, idempotent)
                 → combined_forward_nav (warehouse panel → CombinedBook → ForwardTracker.step, books new days only)
                 → cockpit_forward (ForwardNavRepo upsert, idempotent)
```
Re-materialization is safe: warehouse ingest clears+writes per (symbol, ts-range); the forward tracker
books only dates after the last booked date; the cockpit upserts are idempotent. A backfill re-runs the
graph for a date partition without double-counting.

## Testing strategy

TDD. Dagster assets are directly invocable in-process (call the asset's underlying function with built
resources) — no live cluster needed.
- **Data-source equivalence (THE key test):** the same fixture data, fed through `RawFileSource` and
  `WarehouseSource`, yields the same `CombinedBook.build()` weights + forward NAV within float tolerance
  — proves the SSOT swap is faithful (this is the A→B correctness guarantee).
- **Warehouse panel reader:** ingest a multi-symbol fixture, `read_panel` returns an aligned matrix with
  correct dates/values and NaN where a symbol lacks a bar.
- **Asset wiring:** `combined_forward_nav` materializes against a temp warehouse + temp state and returns
  a forward summary; `cockpit_forward` upserts into a temp SQLite/PG and is idempotent on re-run.
- **Schedule/freshness config:** the `Definitions` load; the schedule has the expected cron; freshness
  policies are attached. (`dagster definitions validate`-style load test.)

## Scope

**In scope (B0):** the 4 assets for the combined-book track; the `CombinedBookSource` port +
`WarehouseSource`/`RawFileSource`; the warehouse panel reader; the Dagster `Definitions` + daily
schedule; the cluster deployment (webserver + daemon + Postgres storage + tailnet) via Argo/Kaniko;
suspend the `fxhnt-forward` CronJob; tests for all of the above.

**Out of scope (later increments):** the other 6 paper tracks (B1); backtest→gauntlet assets (B2);
promote/diversification-gate + live execution (B3); cockpit expansion (B4); sensors/alerting; Dagster
multi-code-location; migrating other consumers off raw files.

## File summary

| File | Change |
|---|---|
| `application/combined_book_source.py` | **create** — `CombinedBookSource` port + `RawFileSource` + `WarehouseSource` |
| `application/combined_book.py` | refactor `__init__`/`_crypto`/`_trend` to use an injected `CombinedBookSource` |
| `adapters/warehouse/duckdb_feature_store.py` | add `read_panel(symbols, feature)` (aligned multi-symbol matrix) |
| `adapters/orchestration/__init__.py`, `assets.py`, `definitions.py` | **create** — Dagster assets + Definitions + daily schedule |
| `config.py` | add `FXHNT_DATA_SOURCE` (warehouse\|raw) for the combined book; Dagster Postgres DSN/db name |
| `pyproject.toml` | add `[orchestration]` extra: `dagster`, `dagster-webserver`, `dagster-postgres` |
| `infra/docker/fxhnt.Dockerfile` | install `.[orchestration]` |
| `infra/k8s/orchestration/` | **create** — dagster-webserver + dagster-daemon Deployments/Services, `dagster.yaml` ConfigMap, tailscale sidecar, netpols |
| `infra/k8s/jobs/fxhnt-forward-cronjob.yaml` | set `suspend: true` (reversible) |
| `infra/argo/orchestration-build-deploy.yaml` | **create** (or extend cockpit build) — Kaniko build + deploy |
| `tests/unit/test_combined_book_source.py` | **create** — RawFileSource vs WarehouseSource equivalence + panel reader |
| `tests/integration/test_orchestration_assets.py` | **create** — asset materialization + cockpit upsert idempotency + Definitions load |
