From decfed966978eaa538cbec66b93f66be50cdd243 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Thu, 25 Jun 2026 23:27:49 +0200 Subject: [PATCH] perf(ingest): crypto warehouse-ingest uses the bulk one-transaction path (was per-symbol serial ~52min) Co-Authored-By: Claude Opus 4.8 --- src/fxhnt/cli.py | 19 ++++++------------- 1 file changed, 6 insertions(+), 13 deletions(-) diff --git a/src/fxhnt/cli.py b/src/fxhnt/cli.py index a40be1b..71c13bb 100644 --- a/src/fxhnt/cli.py +++ b/src/fxhnt/cli.py @@ -347,13 +347,9 @@ def warehouse_ingest( symbol: str = typer.Option("", "--symbol", help="futures: root for /.dbn (required for futures)"), ) -> None: """Bronze -> silver -> gold: normalize raw bars and write them into the warehouse SSOT.""" - import glob import os - import numpy as np - from fxhnt.adapters.warehouse import get_feature_store - from fxhnt.adapters.warehouse.silver_crypto import normalize_crypto from fxhnt.adapters.warehouse.silver_futures import normalize_futures from fxhnt.application.warehouse_ingest import WarehouseIngest @@ -376,15 +372,12 @@ def warehouse_ingest( n = ingest.ingest_bars(symbol, normalize_futures(symbol, days, inst, close, high, low)) typer.echo(f"futures {symbol}: {n} bars -> warehouse") else: # crypto (already validated) - total = 0 - for path in sorted(glob.glob(os.path.join(data_dir, "crypto_pit", "*.npz"))): - sym = os.path.splitext(os.path.basename(path))[0] - with np.load(path) as z: # allow_pickle defaults False — numeric arrays only - high = z["high"] if "high" in z.files else None # real daily H/L when present; - low = z["low"] if "low" in z.files else None # old close-only npz -> high=low=close - bars = normalize_crypto(sym, z["day"], z["close"], high, low) - total += ingest.ingest_bars(sym, bars) - typer.echo(f"crypto: {total} bars -> warehouse") + # Bulk path: read every crypto_pit/*.npz (real H/L when present) and write ALL symbols in ONE + # transaction (one commit, batched multi-row upserts) — the SAME helper the Dagster crypto_bars + # asset uses. Avoids the per-symbol commit overhead of the old serial ingest_bars loop (~52min). + from fxhnt.application.silver_ingest_helpers import ingest_crypto_pit + total = ingest_crypto_pit(store, os.path.join(data_dir, "crypto_pit")) + typer.echo(f"crypto: {total} bars -> warehouse (bulk, one transaction)") finally: store.close()