feat(combined-book): WarehouseSource reads SSOT; equivalence test proves A->B faithful

- 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 <noreply@anthropic.com>

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))

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/<SYM>.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

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."""
+        ...

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()