docs(lifecycle): design spec — B0 Dagster foundation + combined-book via warehouse
Sub-project B increment 0: stand up Dagster on the cluster + migrate the combined-book paper-forward track end-to-end as an asset graph reading the warehouse SSOT (proves A->B), replacing the fxhnt-forward cron. CombinedBookSource port (Warehouse/Raw), 4 assets, cockpit ingest, reversible cutover. B1-B4 (other tracks, gauntlet, promote/live) deferred. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
164
docs/superpowers/specs/2026-06-15-lifecycle-dagster-b0-design.md
Normal file
164
docs/superpowers/specs/2026-06-15-lifecycle-dagster-b0-design.md
Normal file
@@ -0,0 +1,164 @@
|
||||
# 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 |
|
||||
Reference in New Issue
Block a user