From c66bffabef0673fd45a2b0e04097c8a2b9791de8 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 15 Jun 2026 11:16:25 +0200 Subject: [PATCH] feat(combined-book): WarehouseSource reads SSOT; equivalence test proves A->B faithful MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Extend FeatureStore protocol with catalog() and read_panel() so WarehouseSource can be typed against the port (mypy strict clean) - Add silver_ingest_helpers.ingest_crypto_pit — shared bronze→silver→warehouse write path for assets and tests - Implement WarehouseSource in combined_book_source.py; delegates crypto panel to read_panel+_panel and futures to WarehousePriceProvider - Add equivalence test: warehouse-fed panel == raw-file-fed panel (A→B gate) - 291 tests pass, 0 regressions Co-Authored-By: Claude Sonnet 4.6 --- src/fxhnt/application/combined_book_source.py | 27 ++++++++++++++++--- .../application/silver_ingest_helpers.py | 23 ++++++++++++++++ src/fxhnt/ports/warehouse.py | 8 ++++++ .../test_combined_book_source_equivalence.py | 24 +++++++++++++++++ 4 files changed, 79 insertions(+), 3 deletions(-) create mode 100644 src/fxhnt/application/silver_ingest_helpers.py create mode 100644 tests/integration/test_combined_book_source_equivalence.py diff --git a/src/fxhnt/application/combined_book_source.py b/src/fxhnt/application/combined_book_source.py index 4f5f49c..3be2b4e 100644 --- a/src/fxhnt/application/combined_book_source.py +++ b/src/fxhnt/application/combined_book_source.py @@ -1,5 +1,5 @@ """Data source for the combined book — the A→B seam. The book reads its crypto panel + futures series -through this port; RawFileSource is the existing raw-file path, WarehouseSource (next task) reads the SSOT.""" +through this port; RawFileSource is the existing raw-file path, WarehouseSource reads the SSOT.""" from __future__ import annotations from typing import Protocol @@ -7,8 +7,10 @@ from typing import Protocol import numpy as np from fxhnt.adapters.data.dbn_local import LocalDbnLoader -from fxhnt.application.crypto_momentum_research import load_crypto_panel -from fxhnt.domain.models import PriceSeries +from fxhnt.adapters.warehouse.price_provider import WarehousePriceProvider +from fxhnt.application.crypto_momentum_research import _panel, load_crypto_panel +from fxhnt.domain.models import AssetClass, Market, PriceSeries +from fxhnt.ports.warehouse import FeatureStore class CombinedBookSource(Protocol): @@ -32,3 +34,22 @@ class RawFileSource: def futures_series(self, root: str) -> PriceSeries: return self._loader.load(f"{self._dir}/{root}.dbn", root) + + +class WarehouseSource: + """Reads the combined book's data from the warehouse SSOT. Crypto = the panel of all USDT-perp + symbols in the warehouse; futures = per-root continuous front via WarehousePriceProvider.""" + + def __init__(self, store: FeatureStore) -> None: + self._store = store + self._provider = WarehousePriceProvider(store) + + def _crypto_symbols(self) -> list[str]: + return [s for s, *_ in self._store.catalog() if s.endswith("USDT")] + + def crypto_panel(self) -> tuple[list[int], list[str], np.ndarray, np.ndarray]: + data = self._store.read_panel(self._crypto_symbols(), "close") + return _panel(data) + + def futures_series(self, root: str) -> PriceSeries: + return self._provider.fetch(Market(symbol=root, asset_class=AssetClass.FUTURE)) diff --git a/src/fxhnt/application/silver_ingest_helpers.py b/src/fxhnt/application/silver_ingest_helpers.py new file mode 100644 index 0000000..e001aa5 --- /dev/null +++ b/src/fxhnt/application/silver_ingest_helpers.py @@ -0,0 +1,23 @@ +"""Bronze->silver->warehouse ingest helpers for the orchestration assets + tests (one place so the asset +and the equivalence test ingest identically).""" +from __future__ import annotations + +import glob +import os + +import numpy as np + +from fxhnt.adapters.warehouse.silver_crypto import normalize_crypto +from fxhnt.application.warehouse_ingest import WarehouseIngest +from fxhnt.ports.warehouse import FeatureStore + + +def ingest_crypto_pit(store: FeatureStore, crypto_dir: str) -> int: + """Ingest every crypto_pit/.npz (keys day, close) into the warehouse via silver_crypto.""" + ingest = WarehouseIngest(store) + total = 0 + for path in sorted(glob.glob(os.path.join(crypto_dir, "*.npz"))): + sym = os.path.splitext(os.path.basename(path))[0] + with np.load(path) as z: + total += ingest.ingest_bars(sym, normalize_crypto(sym, z["day"], z["close"])) + return total diff --git a/src/fxhnt/ports/warehouse.py b/src/fxhnt/ports/warehouse.py index 5123c09..e718c17 100644 --- a/src/fxhnt/ports/warehouse.py +++ b/src/fxhnt/ports/warehouse.py @@ -24,3 +24,11 @@ class FeatureStore(Protocol): def as_of(self, symbol: str, ts: int, features: list[str] | None = None) -> dict[str, float]: """The latest feature values known at-or-before ts (no lookahead).""" ... + + def catalog(self) -> list[tuple[str, int, int, int]]: + """Per symbol: (symbol, bar_count, first_ts, last_ts) over the 'close' feature.""" + ... + + def read_panel(self, symbols: list[str], feature: str) -> dict[str, dict[int, float]]: + """{symbol: {epoch_day: value}} for one feature across symbols.""" + ... diff --git a/tests/integration/test_combined_book_source_equivalence.py b/tests/integration/test_combined_book_source_equivalence.py new file mode 100644 index 0000000..ae98d1b --- /dev/null +++ b/tests/integration/test_combined_book_source_equivalence.py @@ -0,0 +1,24 @@ +"""WarehouseSource.crypto_panel must equal RawFileSource.crypto_panel on the same data — proves the +SSOT swap is faithful (the A->B guarantee).""" +from __future__ import annotations + +import numpy as np + +from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore +from fxhnt.application.combined_book_source import RawFileSource, WarehouseSource +from fxhnt.application.silver_ingest_helpers import ingest_crypto_pit + + +def test_warehouse_panel_equals_rawfile_panel(tmp_path) -> None: + cp = tmp_path / "crypto_pit"; cp.mkdir() + np.savez(cp / "BTCUSDT.npz", day=np.array([19000, 19001, 19002]), close=np.array([100.0, 110.0, 105.0])) + np.savez(cp / "ETHUSDT.npz", day=np.array([19001, 19002]), close=np.array([50.0, 55.0])) + store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb")) + ingest_crypto_pit(store, str(cp)) + + d_raw, s_raw, c_raw, r_raw = RawFileSource(str(tmp_path)).crypto_panel() + d_wh, s_wh, c_wh, r_wh = WarehouseSource(store).crypto_panel() + assert d_raw == d_wh and s_raw == s_wh + assert np.allclose(c_raw, c_wh, equal_nan=True) + assert np.allclose(r_raw, r_wh, equal_nan=True) + store.close()