gitadmin/fxhnt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(orchestration): assets fetch fresh data before ingest (crypto daily, futures Sunday) — track advances

Browse Source 
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>
This commit is contained in:
jgrusewski
2026-06-15 16:52:25 +02:00
parent f628d78cbb
commit d939f9b33f
1 changed files with 20 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/fxhnt/adapters/orchestration/assets.py
View File

@@ -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: