feat(b1): wire 6 paper-track assets into the Dagster graph

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/fxhnt/adapters/orchestration/assets.py

@@ -120,7 +120,85 @@ def combined_forward_nav(context: AssetExecutionContext) -> dict:  # type: ignor
     }
 
 
-@asset(deps=[combined_forward_nav])
+def _run_paper_tracker(context: AssetExecutionContext, name: str, build_strategy, state_file: str) -> dict:  # type: ignore[type-arg,no-untyped-def]
+    """Best-effort: build the strategy from its live adapter, step the tracker, persist + report state.
+    A transient API error logs a warning and returns the on-disk state without crashing the run."""
+    from fxhnt.application.forward_tracker import ForwardTracker
+
+    path = f"{_data_dir()}/{state_file}.json"
+    try:
+        st = ForwardTracker(build_strategy(), path).step()
+        context.log.info(f"{name}: {st.forward_days}d through {st.last_date}")
+        return {"forward_days": st.forward_days, "last_date": st.last_date}
+    except Exception as e:  # noqa: BLE001 — a transient live-API failure must not kill the nightly run
+        context.log.warning(f"{name} step failed (state unchanged): {e}")
+        return {"forward_days": 0, "last_date": None}
+
+
+@asset
+def sixtyforty_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
+    """Paper-forward 60/40 baseline (Yahoo daily adjusted closes)."""
+    from fxhnt.adapters.data.yahoo_daily import YahooDailyClient
+    from fxhnt.application.paper_strategies import SixtyFortyStrategy
+
+    return _run_paper_tracker(
+        context, "sixtyforty_nav", lambda: SixtyFortyStrategy(YahooDailyClient()), "sixtyforty_state"
+    )
+
+
+@asset
+def multistrat_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
+    """Paper-forward multi-strategy book (Yahoo daily adjusted closes)."""
+    from fxhnt.adapters.data.yahoo_daily import YahooDailyClient
+    from fxhnt.application.paper_strategies import MultiStratStrategy
+
+    return _run_paper_tracker(
+        context, "multistrat_nav", lambda: MultiStratStrategy(YahooDailyClient()), "multistrat_state"
+    )
+
+
+@asset
+def gd_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
+    """Paper-forward growth-discipline book (Yahoo daily adjusted closes)."""
+    from fxhnt.adapters.data.yahoo_daily import YahooDailyClient
+    from fxhnt.application.paper_strategies import GrowthDisciplineStrategy
+
+    return _run_paper_tracker(
+        context, "gd_nav", lambda: GrowthDisciplineStrategy(YahooDailyClient()), "gd_paper_state"
+    )
+
+
+@asset
+def funding_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
+    """Paper-forward crypto funding harvest (Binance public futures API, no key)."""
+    from fxhnt.adapters.data.binance_funding import BinanceFundingClient
+    from fxhnt.application.paper_strategies import FundingCarryStrategy
+
+    return _run_paper_tracker(
+        context, "funding_nav", lambda: FundingCarryStrategy(BinanceFundingClient()), "funding_paper_state"
+    )
+
+
+@asset
+def crossvenue_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
+    """Paper-forward cross-venue funding arb (Binance/Bybit/Hyperliquid public APIs)."""
+    from fxhnt.adapters.data.cross_venue import CrossVenueFundingClient
+    from fxhnt.application.paper_strategies import CrossVenueStrategy
+
+    return _run_paper_tracker(
+        context, "crossvenue_nav", lambda: CrossVenueStrategy(CrossVenueFundingClient()), "crossvenue_state"
+    )
+
+
+@asset
+def poc_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
+    """Paper-forward surfer PoC (crypto residual-momentum + tail-managed VRP; fetches Binance + reads crypto_pit)."""
+    from fxhnt.application.paper_strategies import MomentumVrpStrategy
+
+    return _run_paper_tracker(context, "poc_nav", lambda: MomentumVrpStrategy(), "poc_state")
+
+
+@asset(deps=[combined_forward_nav, sixtyforty_nav, multistrat_nav, gd_nav, funding_nav, crossvenue_nav, poc_nav])
 def cockpit_forward(context: AssetExecutionContext) -> None:
     """Normalize tracker state files and upsert rows + summary into the operational cockpit DB."""
     from fxhnt.application.forward_ingest import ingest_forward_state

src/fxhnt/adapters/orchestration/definitions.py

@@ -1,19 +1,29 @@
 """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
+materializes the track (replaces the fxhnt-forward cron, same 23:30 UTC slot).
 
+NOTE: no `from __future__ import annotations` — Dagster 1.12 runtime type-hint inspection."""
 from dagster import DefaultScheduleStatus, Definitions, ScheduleDefinition, define_asset_job
 
 from fxhnt.adapters.orchestration.assets import (
     cockpit_forward,
     combined_forward_nav,
+    crossvenue_nav,
     crypto_bars,
+    funding_nav,
     futures_bars,
+    gd_nav,
+    multistrat_nav,
+    poc_nav,
+    sixtyforty_nav,
 )
 
 combined_book_job = define_asset_job(
     name="combined_book_forward_job",
-    selection=[crypto_bars, futures_bars, combined_forward_nav, cockpit_forward],
+    selection=[
+        crypto_bars, futures_bars, combined_forward_nav,
+        sixtyforty_nav, multistrat_nav, gd_nav, funding_nav, crossvenue_nav, poc_nav,
+        cockpit_forward,
+    ],
 )
 
 daily_combined_book_schedule = ScheduleDefinition(
@@ -25,7 +35,11 @@ daily_combined_book_schedule = ScheduleDefinition(
 )
 
 defs = Definitions(
-    assets=[crypto_bars, futures_bars, combined_forward_nav, cockpit_forward],
+    assets=[
+        crypto_bars, futures_bars, combined_forward_nav,
+        sixtyforty_nav, multistrat_nav, gd_nav, funding_nav, crossvenue_nav, poc_nav,
+        cockpit_forward,
+    ],
     jobs=[combined_book_job],
     schedules=[daily_combined_book_schedule],
 )

tests/integration/test_orchestration_definitions.py

@@ -11,6 +11,9 @@ def test_definitions_load_with_assets_and_schedule() -> None:
     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
+    # B1: the six paper-track assets are wired into the graph alongside the B0 four
+    paper = {"sixtyforty_nav", "multistrat_nav", "gd_nav", "funding_nav", "crossvenue_nav", "poc_nav"}
+    assert paper <= asset_keys, f"missing paper-track assets: {paper - asset_keys}"
 
     # --- schedule present with the right cron ---
     # defs.schedules is a list[ScheduleDefinition] (or None when empty)
@@ -22,3 +25,11 @@ def test_definitions_load_with_assets_and_schedule() -> None:
     from dagster import DefaultScheduleStatus
     daily = next(s for s in scheds if getattr(s, "cron_schedule", "") == "30 23 * * *")
     assert daily.default_status == DefaultScheduleStatus.RUNNING
+
+    # --- cockpit_forward depends on every paper-track asset (so it ingests their state files) ---
+    from dagster import AssetKey
+    cockpit_def = repo.assets_defs_by_key[AssetKey("cockpit_forward")]
+    deps = cockpit_def.asset_deps[AssetKey("cockpit_forward")]
+    upstream = {k.to_user_string() for k in deps}
+    expected_upstream = {"combined_forward_nav"} | paper
+    assert expected_upstream <= upstream, f"cockpit_forward missing upstream: {expected_upstream - upstream}"