diff --git a/src/fxhnt/adapters/orchestration/assets.py b/src/fxhnt/adapters/orchestration/assets.py index 2fe610f..2e9ab27 100644 --- a/src/fxhnt/adapters/orchestration/assets.py +++ b/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 diff --git a/src/fxhnt/adapters/orchestration/definitions.py b/src/fxhnt/adapters/orchestration/definitions.py index 79f1eb8..13c1cf2 100644 --- a/src/fxhnt/adapters/orchestration/definitions.py +++ b/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], ) diff --git a/tests/integration/test_orchestration_definitions.py b/tests/integration/test_orchestration_definitions.py index 6ee020c..ae5187f 100644 --- a/tests/integration/test_orchestration_definitions.py +++ b/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}"