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

perf(ingest): crypto warehouse-ingest uses the bulk one-transaction path (was per-symbol serial ~52min)

Browse Source 
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-06-25 23:27:49 +02:00
parent 46d2d4abe8
commit decfed9669
1 changed files with 6 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/fxhnt/cli.py
View File

@@ -347,13 +347,9 @@ def warehouse_ingest(
symbol: str = typer.Option("", "--symbol", help="futures: root for <data_dir>/<symbol>.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()