# Unified Data Layer (Warehouse-SSOT) — Design Spec

**Date:** 2026-06-15
**Status:** Approved (design)
**Sub-project:** A of the "transform fxhnt into a research platform" effort. Sub-projects B (strategy lifecycle), C (research environment), D (cockpit expansion) are separate later specs that build on this foundation.

## Goal

Make the existing warehouse the **single source of truth (SSOT)** for market data: one point-in-time
store that fetchers write to and that research, backtest and forward all read from — replacing the
current state where data lives in three parallel homes and the designed SSOT sits orphaned.

This spec covers **increment 1**: daily bars (futures + crypto) flowing bronze→silver→gold into the
warehouse, consumed through the existing `DataProvider` port, plus a catalog/freshness view. Funding
rates, retiring the other homes, and migrating the remaining consumers are explicit later increments.

## Context — why the SSOT is orphaned

`ports/warehouse.py` already declares the SSOT: *"the gold-layer feature store of the SSOT
architecture. Point-in-time by construction… the discipline the analyst agents stand on."* There is an
adapter (`adapters/warehouse/duckdb_feature_store.py`) and a write pipeline
(`application/feature_pipeline.py`). **But nothing reads from it** — a grep for `as_of`/`read_features`
in research/backtest/factory finds only the pipeline's own import.

The reason: the warehouse was built for **MBP-10 microstructure features** — `FeaturePipeline` consumes
`Mbp10Snapshot` streams through microprice/OFI calculators. But research/backtest/forward run on
**daily OHLCV bars + funding**. The warehouse expected the wrong *shape* of data, so it was never wired
in. Meanwhile consumers read from three other places:

| Data home | What | Who reads | Point-in-time? |
|---|---|---|---|
| DuckDB `AnalyticalStore` (`prices`/`returns`) | adjusted bars + return series | fleet, research, forward, portfolio_eval, runtime | ❌ "latest series" |
| Raw files (`crypto_pit/`, `*.dbn`) | crypto bars + futures bars | combined_book, carry/tsmom/vrp research, fetchers | ❌ files |
| DuckDB `FeatureStore` (warehouse) | the intended SSOT, `as_of` | nobody (orphaned) | ✅ by construction |
| foxhunt surfer scripts | own fetches + local JSON | the 6 laptop crons | ❌ separate island |

## Key insight — the long-format schema already generalizes

The warehouse table is `features(symbol, ts, feature, value)` (long format), read point-in-time via
`as_of(symbol, ts)` = latest value of each feature at-or-before `ts`. This generalizes cleanly beyond
microstructure:

- A **daily bar** = features `{open, high, low, close, volume}` at `ts` = the bar day.
- A **funding rate** (later increment) = feature `{funding_rate}` at `ts`.
- Microstructure features (existing) coexist in the same table, same symbol namespace.

So converging on the SSOT does not require a new store or schema — it requires **silver normalizers**
that turn raw bars into canonical rows, and a **price provider** that reads bars back out as a
`PriceSeries`. One store then serves microstructure, bars and funding uniformly.

## Design

### Canonical conventions

- **Timestamp:** epoch **seconds, UTC, at 00:00:00 of the bar day**. Silver normalizers convert each
  source's native unit to this (dbn epoch-days → ×86400; binance epoch-ms → ÷1000 then floor to the
  UTC day). This is the single `ts` convention for daily bars in the warehouse.
- **Bar features:** the canonical feature names for a daily bar are exactly `open`, `high`, `low`,
  `close`, `volume`. `close` is the adjusted/continuous close (matching today's `PriceSeries.close`);
  `high`/`low` follow the same adjustment as in the existing adapters (Task-9 ratio-adjusted for the
  continuous futures front; Task-8 ratio-scaled for adjusted equities — not in scope here but the
  convention is shared).
- **Symbol:** the warehouse `symbol` is the market root string (e.g. `ES`, `BTCUSDT`). Asset class is
  not part of the warehouse key in increment 1 (bars are keyed by symbol+ts); the `WarehousePriceProvider`
  is constructed for a known `Market` so it supplies asset_class on read.

### Component 1 — `BarRecord` + silver normalizers

A small canonical value object and one normalizer per raw source.

`domain/warehouse/bar.py`:
```python
@dataclass(frozen=True)
class BarRecord:
    symbol: str
    ts: int                       # epoch seconds, 00:00:00 UTC of the bar day
    open: float
    high: float
    low: float
    close: float
    volume: float

    def as_feature_row(self) -> tuple[int, dict[str, float]]:
        return (self.ts, {"open": self.open, "high": self.high, "low": self.low,
                          "close": self.close, "volume": self.volume})
```

`adapters/warehouse/silver_futures.py` — `normalize_futures(symbol, days, inst, o, h, l, c, vol) ->
list[BarRecord]`: reuses the continuous-front ratio-adjustment from `dbn_local._front_to_series`
(roll books 0 return; high/low scaled by the same per-day factor as close) and converts epoch-days →
epoch-seconds. (The shared adjustment math should be factored so `dbn_local` and the silver normalizer
call ONE implementation — no duplication.)

`adapters/warehouse/silver_crypto.py` — `normalize_crypto(symbol, klines) -> list[BarRecord]`: maps
Binance daily klines (open_time ms, OHLCV) to `BarRecord` with ts floored to the UTC day. Survivorship
handling stays in the existing `crypto_fetch` (bronze); the normalizer only reshapes.

Each normalizer is a pure function over already-fetched raw data (the bronze fetchers stay as they are),
so both are unit-testable on tiny fixtures with no network.

### Component 2 — gold ingest (`WarehouseIngest` + CLI)

`application/warehouse_ingest.py`:
```python
class WarehouseIngest:
    def __init__(self, store: FeatureStore) -> None: self._store = store
    def ingest_bars(self, symbol: str, bars: list[BarRecord]) -> int:
        self._store.write_features(symbol, [b.as_feature_row() for b in bars])
        return len(bars)
```

CLI `fxhnt warehouse-ingest --data-dir <dir> --source futures|crypto [--symbol <S>]`: reads the bronze
data in `data_dir`, runs the matching silver normalizer, writes to the warehouse via `WarehouseIngest`.
Writes are idempotent (the feature store already clears the (symbol, ts-range) before insert), so
re-ingest is safe. Echoes per-symbol counts.

### Component 3 — `WarehousePriceProvider` (the SSOT read path)

`adapters/warehouse/price_provider.py` — implements the existing `DataProvider` Protocol so every
current consumer (research, forward) reads the SSOT with no change to backtest/gauntlet:
```python
class WarehousePriceProvider:
    name = "warehouse"
    def __init__(self, store: FeatureStore) -> None: self._store = store
    def fetch(self, market: Market, start: str | None = None, end: str | None = None) -> PriceSeries:
        rows = self._store.read_features(market.symbol, start_ts=_iso_to_ts(start),
                                         end_ts=_iso_to_ts(end), features=["open","high","low","close","volume"])
        # rows are ts-ascending; build aligned arrays, dates = ISO day from ts
        ...
        return PriceSeries(market=market, dates=dates, close=close, high=high, low=low)
```
- Bars are read via `read_features` (a bounded range read), already ts-ordered.
- Rows missing `high`/`low` fall back to `close` (PriceSeries default already handles None, but the
  provider passes explicit arrays).
- Wiring: a config flag `FXHNT_DATA_SOURCE=warehouse|provider` (default `provider` for now) selects
  whether the composition root injects `WarehousePriceProvider` or the existing fetch/AnalyticalStore
  path into research/forward. This makes the cutover reversible and testable side-by-side.

### Component 4 — catalog / freshness

`application/warehouse_catalog.py` + CLI `fxhnt warehouse-catalog`: queries the feature store for, per
symbol, the bar count, min/max `ts` (date range), and the max `ts` of the `close` feature (freshness).
Output is a small table: `symbol | bars | first | last`. This is the "organized" win — you can see at a
glance what data the platform holds and whether it is stale.

A read-only helper on the store is acceptable (e.g. a `catalog()` method on `DuckDbFeatureStore`
returning rows), or a direct query in the catalog service; the implementer picks the simpler one that
keeps the SSOT the only data home.

### Config

Add to `Settings` (`config.py`): `warehouse_path: str = Field(default=str(_DATA_DIR / "warehouse.duckdb"))`
and `data_source: str = Field(default="provider")` (the cutover flag). `ensure_dirs()` creates the
warehouse parent dir. `warehouse.duckdb` is separate from `analytical.duckdb`; retiring the latter is a
later increment.

## Data flow + point-in-time guarantee

```
fetch (bronze: .dbn / crypto_pit)
  → normalize (silver: BarRecord stream, canonical ts)
  → feature store (gold: features table, idempotent)
  → WarehousePriceProvider.fetch  (range read of OHLCV features)
  → PriceSeries → run_backtest (unchanged)
```

Lookahead is structurally impossible on the read path: bars are read with an upper `end_ts` bound and,
for as-of-date research, via `as_of(ts)` which returns only features ≤ ts. The silver→gold transform is
deterministic and causal (same SSOT input → same gold), matching the existing `FeaturePipeline`
contract.

## Scope

**In scope (increment 1):**
- `BarRecord` + futures/crypto silver normalizers (shared adjustment math, no duplication with dbn_local).
- `WarehouseIngest` service + `warehouse-ingest` CLI.
- `WarehousePriceProvider` implementing `DataProvider`, selectable via `FXHNT_DATA_SOURCE`.
- `warehouse-catalog` CLI + the supporting read.
- `warehouse_path` + `data_source` config.
- Tests for all of the above.

**Out of scope (explicit later increments, each its own spec):**
- Funding-rate ingestion into the warehouse (next increment — high value for the funding harvest).
- Migrating `combined_book` / `carry` / `vrp` / `tsmom` research off raw-file reads.
- Retiring the DuckDB `AnalyticalStore` `prices` table (and merging into the warehouse).
- Bringing the foxhunt surfer crons onto the warehouse.
- Wiring `as_of` into the factory/analyst feature reads.
- Equities adjustment in the silver layer (futures + crypto only here).

## Testing strategy

TDD per task. Everything is unit-testable without network: bronze fetchers are untouched, normalizers
are pure functions over fixture data, and the warehouse is an embedded DuckDB temp file.

- **silver_futures:** a tiny days/inst/OHLCV fixture (incl. a roll) → `BarRecord`s with correct
  epoch-second ts and roll-consistent ratio-adjusted high/low (mirrors the dbn_local Task-9 test).
- **silver_crypto:** klines fixture → `BarRecord`s with ts floored to the UTC day, OHLCV mapped.
- **ingest round-trip:** ingest a fixture, then `WarehousePriceProvider.fetch` returns a `PriceSeries`
  equal (within float tolerance) to building it directly from the same fixture — proves SSOT equivalence.
- **point-in-time:** after ingesting bars through day T, `as_of(symbol, ts<T)` never returns a bar > ts.
- **catalog:** ranges/counts/freshness match the ingested fixture.
- **idempotency:** ingesting the same fixture twice yields one set of rows (no duplication).

## File summary

| File | Change |
|---|---|
| `domain/warehouse/bar.py` | **create** — `BarRecord` value object |
| `adapters/warehouse/silver_futures.py` | **create** — futures bronze→silver normalizer (shared adj math) |
| `adapters/warehouse/silver_crypto.py` | **create** — crypto bronze→silver normalizer |
| `adapters/data/dbn_local.py` | factor out the ratio-adjust math so silver_futures reuses it (no dup) |
| `application/warehouse_ingest.py` | **create** — `WarehouseIngest` service |
| `adapters/warehouse/price_provider.py` | **create** — `WarehousePriceProvider` (DataProvider) |
| `application/warehouse_catalog.py` | **create** — catalog/freshness read |
| `adapters/warehouse/duckdb_feature_store.py` | add a `catalog()` read (if chosen) |
| `config.py` | add `warehouse_path`, `data_source`; `ensure_dirs` covers warehouse parent |
| `cli.py` | add `warehouse-ingest` + `warehouse-catalog` commands; honor `data_source` in composition |
| `tests/unit/test_warehouse_silver.py` | **create** — normalizer tests |
| `tests/unit/test_warehouse_price_provider.py` | **create** — round-trip + point-in-time tests |
| `tests/integration/test_warehouse_ingest_catalog.py` | **create** — ingest + catalog + idempotency |
