From d939f9b33f0116538e6506bf28cb6ec746862eb3 Mon Sep 17 00:00:00 2001
From: jgrusewski <jgrusewski@localhost>
Date: Mon, 15 Jun 2026 16:52:25 +0200
Subject: [PATCH] =?UTF-8?q?fix(orchestration):=20assets=20fetch=20fresh=20?=
 =?UTF-8?q?data=20before=20ingest=20(crypto=20daily,=20futures=20Sunday)?=
 =?UTF-8?q?=20=E2=80=94=20track=20advances?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The B0 assets only ingested existing files → forward track froze on stale data. Mirror the retired
cron: crypto_bars fetches crypto_pit (Binance, free, daily, best-effort); futures_bars fetches via
Databento on Sundays (cost-capped, key-gated) before the silver→warehouse ingest.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/fxhnt/adapters/orchestration/assets.py | 21 ++++++++++++++++++++-
 1 file changed, 20 insertions(+), 1 deletion(-)

diff --git a/src/fxhnt/adapters/orchestration/assets.py b/src/fxhnt/adapters/orchestration/assets.py
index 02e1d39..2fe610f 100644
--- a/src/fxhnt/adapters/orchestration/assets.py
+++ b/src/fxhnt/adapters/orchestration/assets.py
@@ -39,13 +39,20 @@ def build_combined_forward_nav(
 
 @asset
 def crypto_bars(context: AssetExecutionContext) -> int:
-    """Bronze → silver → warehouse: ingest all crypto_pit NPZ files."""
+    """Bronze FETCH (Binance, free, daily) → silver → warehouse: refresh + ingest the crypto_pit panel.
+    Without the fetch the forward track would ingest stale data and never advance."""
+    from fxhnt.adapters.data.crypto_fetch import fetch_crypto_pit
     from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
     from fxhnt.application.silver_ingest_helpers import ingest_crypto_pit
     from fxhnt.config import get_settings
 
     s = get_settings()
     s.ensure_dirs()
+    try:                                  # best-effort: a transient Binance hiccup must not kill the nightly run
+        fetched, total = fetch_crypto_pit(_data_dir())
+        context.log.info(f"crypto_bars fetched {fetched}/{total} symbols")
+    except Exception as e:  # noqa: BLE001
+        context.log.warning(f"crypto_bars fetch failed (ingesting existing data): {e}")
     n = ingest_crypto_pit(DuckDbFeatureStore(s.warehouse_path), f"{_data_dir()}/crypto_pit")
     context.log.info(f"crypto_bars ingested {n} rows")
     return n
@@ -57,6 +64,9 @@ def futures_bars(context: AssetExecutionContext) -> int:
 
     Skips (with a warning) any root whose DBN file is missing — never fails the asset.
     """
+    import datetime as dt
+    import os
+
     from fxhnt.adapters.data.dbn_local import read_front_arrays
     from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
     from fxhnt.adapters.warehouse.silver_futures import normalize_futures
@@ -65,6 +75,15 @@ def futures_bars(context: AssetExecutionContext) -> int:
 
     s = get_settings()
     s.ensure_dirs()
+    # Futures fetch is Databento (paid) — refresh only on Sundays (UTC), cost-capped, like the retired cron.
+    if dt.datetime.now(dt.timezone.utc).isoweekday() == 7 and os.environ.get("DATABENTO_API_KEY"):
+        from fxhnt.adapters.data.futures_fetch import fetch_futures
+        start = (dt.date.today() - dt.timedelta(days=365 * s.databento.fetch_years)).isoformat()
+        try:
+            recs, ok, spent = fetch_futures(_data_dir(), start=start, max_cost_usd=s.databento.max_cost_usd)
+            context.log.info(f"futures_bars fetched {ok} roots, {recs} records, ${spent:.2f} (since {start})")
+        except Exception as e:  # noqa: BLE001
+            context.log.warning(f"futures_bars fetch failed (ingesting existing data): {e}")
     ingest = WarehouseIngest(DuckDbFeatureStore(s.warehouse_path))
     total = 0
     for root in _FUTURES: