From f4cfec0f73d64eecef42daa5bd042c1016e1b507 Mon Sep 17 00:00:00 2001
From: jgrusewski <jgrusewski@localhost>
Date: Mon, 15 Jun 2026 11:51:55 +0200
Subject: [PATCH] feat(orchestration): Definitions + daily combined-book
 schedule

Wire the 4 B0 assets into combined_book_forward_job and a
ScheduleDefinition (cron 30 23 * * *, UTC) via a Dagster Definitions
object; dagster definitions validate passes.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 .../adapters/orchestration/definitions.py     | 30 +++++++++++++++++++
 .../test_orchestration_definitions.py         | 20 +++++++++++++
 2 files changed, 50 insertions(+)
 create mode 100644 src/fxhnt/adapters/orchestration/definitions.py
 create mode 100644 tests/integration/test_orchestration_definitions.py

diff --git a/src/fxhnt/adapters/orchestration/definitions.py b/src/fxhnt/adapters/orchestration/definitions.py
new file mode 100644
index 0000000..da4fbaf
--- /dev/null
+++ b/src/fxhnt/adapters/orchestration/definitions.py
@@ -0,0 +1,30 @@
+"""Dagster Definitions for the B0 lifecycle graph — the 4 combined-book assets + a daily schedule that
+materializes the track (replaces the fxhnt-forward cron, same 23:30 UTC slot)."""
+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 * * *",
+    execution_timezone="UTC",
+)
+
+defs = Definitions(
+    assets=[crypto_bars, futures_bars, combined_forward_nav, cockpit_forward],
+    jobs=[combined_book_job],
+    schedules=[daily_combined_book_schedule],
+)
diff --git a/tests/integration/test_orchestration_definitions.py b/tests/integration/test_orchestration_definitions.py
new file mode 100644
index 0000000..5ac0d3b
--- /dev/null
+++ b/tests/integration/test_orchestration_definitions.py
@@ -0,0 +1,20 @@
+"""The Definitions load and expose a daily schedule over the combined-book asset job + the 4 assets."""
+from __future__ import annotations
+
+
+def test_definitions_load_with_assets_and_schedule() -> None:
+    from fxhnt.adapters.orchestration.definitions import defs
+
+    # --- assets present ---
+    # defs.get_repository_def() is available in Dagster 1.12; its .assets_defs_by_key
+    # maps AssetKey → AssetsDefinition for every registered asset.
+    repo = defs.get_repository_def()
+    asset_keys = {k.to_user_string() for k in repo.assets_defs_by_key}
+    assert {"crypto_bars", "futures_bars", "combined_forward_nav", "cockpit_forward"} <= asset_keys
+
+    # --- schedule present with the right cron ---
+    # defs.schedules is a list[ScheduleDefinition] (or None when empty)
+    scheds = list(defs.schedules or [])
+    assert any(getattr(s, "cron_schedule", "") == "30 23 * * *" for s in scheds), (
+        f"expected a schedule with cron '30 23 * * *'; got {[getattr(s, 'cron_schedule', None) for s in scheds]}"
+    )