From 9878207ecd9a4760f27bf78020892453a262b49c Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 15 Jun 2026 10:40:42 +0200 Subject: [PATCH] =?UTF-8?q?docs(lifecycle):=20design=20spec=20=E2=80=94=20?= =?UTF-8?q?B0=20Dagster=20foundation=20+=20combined-book=20via=20warehouse?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../2026-06-15-lifecycle-dagster-b0-design.md | 164 ++++++++++++++++++ 1 file changed, 164 insertions(+) create mode 100644 docs/superpowers/specs/2026-06-15-lifecycle-dagster-b0-design.md diff --git a/docs/superpowers/specs/2026-06-15-lifecycle-dagster-b0-design.md b/docs/superpowers/specs/2026-06-15-lifecycle-dagster-b0-design.md new file mode 100644 index 0000000..ebaadfe --- /dev/null +++ b/docs/superpowers/specs/2026-06-15-lifecycle-dagster-b0-design.md @@ -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(.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/.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 |