# Unified Data Layer (Warehouse-SSOT) — Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Make the existing point-in-time warehouse the single source of truth for daily bars — futures + crypto flowing bronze→silver→gold, consumed through the existing `DataProvider` port, plus a catalog/freshness view.

**Architecture:** The warehouse table `features(symbol, ts, feature, value)` (long format, `as_of` point-in-time) already generalizes to bars: a daily bar = features `{close, high, low}` at `ts`. Silver normalizers turn raw futures (.dbn continuous front) and crypto (`crypto_pit/*.npz`) into canonical `BarRecord`s; a `WarehouseIngest` service writes them to the gold store; a `WarehousePriceProvider` reads them back as a `PriceSeries` so research/backtest consume the SSOT unchanged.

**Tech Stack:** Python 3.11, numpy, DuckDB (embedded), pydantic-settings, typer (CLI), pytest. Tests: `python3 -m pytest <path> -q` (pythonpath=`src`). `mypy strict = true`.

**Reference spec:** `docs/superpowers/specs/2026-06-15-unified-data-layer-design.md`

**Scope note (narrowing vs spec, intentional):** Bronze reality — `crypto_pit/*.npz` is close-only (keys `day, close, qvol, funding`) and the futures front-selection carries no `open`. So increment-1 bars carry **`{close, high, low}`** only (crypto: `high=low=close`). `open` and `volume` are deferred (not available from bronze / not consumed by the provider). Funding, retiring the other data homes, and migrating the remaining consumers stay deferred per the spec.

**Conventions:**
- Canonical `ts` = **epoch seconds, UTC, 00:00:00 of the bar day**. Both bronze sources expose epoch-**days** (futures `df["ts_event"]//(86400*1e9)`; crypto `kd = open_time_ms // DAY_MS`), so `ts = int(day) * 86_400`.
- Tests live in `tests/unit/` (pure) and `tests/integration/` (DuckDB temp file), matching the repo layout.

---

## Task 1: Config — `warehouse_path`

**Files:**
- Modify: `src/fxhnt/config.py:99-119`
- Test: `tests/unit/test_config_warehouse.py`

- [ ] **Step 1: Write the failing test**

```python
# tests/unit/test_config_warehouse.py
"""Settings expose a warehouse_path and ensure_dirs creates its parent."""
from __future__ import annotations

from pathlib import Path

from fxhnt.config import Settings


def test_settings_has_warehouse_path_default() -> None:
    s = Settings()
    assert s.warehouse_path.endswith("warehouse.duckdb")


def test_ensure_dirs_creates_warehouse_parent(tmp_path) -> None:
    target = tmp_path / "wh" / "warehouse.duckdb"
    s = Settings(warehouse_path=str(target))
    s.ensure_dirs()
    assert target.parent.is_dir()
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/unit/test_config_warehouse.py -q`
Expected: FAIL — `Settings` has no `warehouse_path`.

- [ ] **Step 3: Implement**

In `src/fxhnt/config.py`, add the field after `analytical_path` (line 106) and extend `ensure_dirs`:

```python
    analytical_path: str = Field(default=str(_DATA_DIR / "analytical.duckdb"))
    # Warehouse (point-in-time SSOT: bars, funding, microstructure features) — DuckDB embedded file.
    warehouse_path: str = Field(default=str(_DATA_DIR / "warehouse.duckdb"))
```

And in `ensure_dirs`:

```python
    def ensure_dirs(self) -> None:
        _DATA_DIR.mkdir(parents=True, exist_ok=True)
        Path(self.analytical_path).parent.mkdir(parents=True, exist_ok=True)
        Path(self.warehouse_path).parent.mkdir(parents=True, exist_ok=True)
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest tests/unit/test_config_warehouse.py -q`
Expected: PASS (2 passed).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/config.py tests/unit/test_config_warehouse.py
git commit -m "feat(config): add warehouse_path setting"
```

---

## Task 2: `BarRecord` value object

**Files:**
- Create: `src/fxhnt/domain/warehouse/__init__.py`, `src/fxhnt/domain/warehouse/bar.py`
- Test: `tests/unit/test_bar_record.py`

- [ ] **Step 1: Write the failing test**

```python
# tests/unit/test_bar_record.py
"""BarRecord is a canonical daily bar that serializes to a warehouse feature row."""
from __future__ import annotations

from fxhnt.domain.warehouse.bar import BarRecord


def test_bar_record_as_feature_row() -> None:
    bar = BarRecord(symbol="ES", ts=1_577_836_800, close=100.0, high=101.0, low=99.0)
    ts, feats = bar.as_feature_row()
    assert ts == 1_577_836_800
    assert feats == {"close": 100.0, "high": 101.0, "low": 99.0}
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/unit/test_bar_record.py -q`
Expected: FAIL — `ModuleNotFoundError: fxhnt.domain.warehouse.bar`.

- [ ] **Step 3: Implement**

Create `src/fxhnt/domain/warehouse/__init__.py` (empty):

```python
"""Warehouse domain value objects (canonical records that flow into the SSOT)."""
```

Create `src/fxhnt/domain/warehouse/bar.py`:

```python
"""BarRecord — a canonical daily bar (close/high/low) keyed by epoch-second ts, written to the
warehouse as point-in-time feature rows. open/volume are omitted: neither bronze source supplies a
real open, and nothing consumes volume in this increment."""
from __future__ import annotations

from dataclasses import dataclass


@dataclass(frozen=True)
class BarRecord:
    symbol: str
    ts: int                    # epoch seconds, 00:00:00 UTC of the bar day
    close: float
    high: float
    low: float

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

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest tests/unit/test_bar_record.py -q`
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/domain/warehouse/__init__.py src/fxhnt/domain/warehouse/bar.py tests/unit/test_bar_record.py
git commit -m "feat(warehouse): BarRecord canonical daily-bar value object"
```

---

## Task 3: Factor dbn_local into reusable pieces (no duplication)

**Files:**
- Modify: `src/fxhnt/adapters/data/dbn_local.py:22-70`
- Test: `tests/unit/test_dbn_continuous_front.py` (new) + existing `tests/unit/test_adapters_ohlc.py` must still pass

- [ ] **Step 1: Write the failing test**

```python
# tests/unit/test_dbn_continuous_front.py
"""continuous_front_adjust is the pure ratio-back-adjustment shared by the loader and the silver layer."""
from __future__ import annotations

import numpy as np

from fxhnt.adapters.data.dbn_local import continuous_front_adjust


def test_continuous_front_adjust_books_zero_return_on_roll() -> None:
    inst = np.array([10, 10, 20], dtype=np.int64)      # roll between day 1 and day 2
    close = np.array([100.0, 102.0, 50.0])             # day 2 is a different contract level
    high = np.array([101.0, 103.0, 51.0])
    low = np.array([99.0, 101.0, 49.0])
    syn_close, syn_high, syn_low = continuous_front_adjust(inst, close, high, low)
    assert syn_close[2] == syn_close[1]                # roll -> flat synthetic level (no gap)
    assert syn_high[2] / syn_close[2] == 51.0 / 50.0   # intraday range preserved proportionally
    assert syn_low[2] / syn_close[2] == 49.0 / 50.0
    assert syn_high[0] >= syn_close[0] >= syn_low[0]
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/unit/test_dbn_continuous_front.py -q`
Expected: FAIL — `cannot import name 'continuous_front_adjust'`.

- [ ] **Step 3: Implement the refactor**

In `src/fxhnt/adapters/data/dbn_local.py`, replace the `_front_to_series` function (lines 22-43) and the body of `LocalDbnLoader.load` (lines 49-70) with this structure — a pure adjustment, a pure-ish DBN reader, and a thin composition. Behavior is unchanged.

```python
def continuous_front_adjust(inst: np.ndarray, close: np.ndarray, high: np.ndarray,
                            low: np.ndarray) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Ratio back-adjust a front-month series: roll days (instrument_id change) book a 0 return so
    there is no cross-contract gap; high/low are scaled by the same per-day factor as close, keeping the
    intraday range proportional on the adjusted scale. cumprod of ratios (not exp(cumsum(log))) keeps
    roll days bit-for-bit flat. Returns (syn_close, syn_high, syn_low)."""
    same_contract = np.zeros(len(close), bool)
    same_contract[1:] = inst[1:] == inst[:-1]
    ratio = np.ones_like(close)
    ratio[1:] = np.where(same_contract[1:], close[1:] / close[:-1], 1.0)
    synthetic = 100.0 * np.cumprod(ratio)
    safe_high = np.where(np.isfinite(high), high, close)
    safe_low = np.where(np.isfinite(low), low, close)
    syn_high = np.divide(safe_high, close, out=np.ones_like(synthetic), where=close > 0) * synthetic
    syn_low = np.divide(safe_low, close, out=np.ones_like(synthetic), where=close > 0) * synthetic
    return synthetic, syn_high, syn_low


def read_front_arrays(path: str, symbol: str) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
    """Read a Databento OHLCV-1d .dbn and return the raw continuous-front arrays
    (days, instrument_id, close, high, low) — outrights only, front = highest-volume per day."""
    import databento as db

    df = db.DBNStore.from_file(path).to_df().reset_index()
    if df.empty or "close" not in df.columns:
        raise ValueError(f"{path}: no OHLCV close")
    df = df[df["close"] > 0]
    pattern = re.compile("^" + re.escape(symbol) + "[" + _MONTHS + r"]\d{1,2}$")
    df = df[df["symbol"].astype(str).str.match(pattern)]
    if df.empty:
        raise ValueError(f"{path}: no outright contracts match {symbol}")
    df = df.copy()
    df["day"] = df["ts_event"].astype("int64") // (86_400 * 10**9)
    front = df.loc[df.groupby("day")["volume"].idxmax(),
                   ["day", "instrument_id", "close", "high", "low"]].sort_values("day")
    return (front["day"].to_numpy(np.int64), front["instrument_id"].to_numpy(np.int64),
            front["close"].to_numpy(np.float64), front["high"].to_numpy(np.float64),
            front["low"].to_numpy(np.float64))


def _front_to_series(symbol: str, asset_class: AssetClass, days: np.ndarray, inst: np.ndarray,
                     close: np.ndarray, high: np.ndarray, low: np.ndarray) -> PriceSeries:
    """Build the continuous front-month PriceSeries from raw front arrays."""
    synthetic, syn_high, syn_low = continuous_front_adjust(inst, close, high, low)
    dates = tuple(dt.datetime.fromtimestamp(int(d) * 86_400, dt.timezone.utc).strftime("%Y-%m-%d")
                  for d in days)
    return PriceSeries(market=Market(symbol=symbol, asset_class=asset_class), dates=dates,
                       close=synthetic, high=syn_high, low=syn_low)


class LocalDbnLoader:
    name = "dbn_local"

    def load(self, path: str, symbol: str, asset_class: AssetClass = AssetClass.FUTURE) -> PriceSeries:
        return _front_to_series(symbol, asset_class, *read_front_arrays(path, symbol))
```

- [ ] **Step 4: Run tests to verify they pass**

Run: `python3 -m pytest tests/unit/test_dbn_continuous_front.py tests/unit/test_adapters_ohlc.py -q`
Expected: PASS — the new test plus the existing `test_dbn_local_front_to_series_preserves_range_across_roll` (which calls `_front_to_series`, now delegating to `continuous_front_adjust`).

- [ ] **Step 5: Verify full suite + mypy**

Run: `python3 -m pytest tests/ -q` then `python3 -m mypy src/fxhnt/adapters/data/dbn_local.py`
Expected: PASS / `Success: no issues found`.

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/adapters/data/dbn_local.py tests/unit/test_dbn_continuous_front.py
git commit -m "refactor(dbn_local): extract continuous_front_adjust + read_front_arrays for reuse"
```

---

## Task 4: Silver normalizer — futures

**Files:**
- Create: `src/fxhnt/adapters/warehouse/silver_futures.py`
- Test: `tests/unit/test_warehouse_silver.py`

- [ ] **Step 1: Write the failing test**

```python
# tests/unit/test_warehouse_silver.py
"""Silver normalizers: raw bronze arrays -> canonical BarRecords (epoch-second ts)."""
from __future__ import annotations

import numpy as np


def test_normalize_futures_builds_adjusted_bars_with_epoch_second_ts() -> None:
    from fxhnt.adapters.warehouse.silver_futures import normalize_futures
    days = np.array([0, 1, 2], dtype=np.int64)        # epoch-days
    inst = np.array([10, 10, 20], dtype=np.int64)     # roll on day 2
    close = np.array([100.0, 102.0, 50.0])
    high = np.array([101.0, 103.0, 51.0])
    low = np.array([99.0, 101.0, 49.0])
    bars = normalize_futures("ES", days, inst, close, high, low)
    assert [b.ts for b in bars] == [0, 86_400, 172_800]          # day * 86400
    assert bars[2].close == bars[1].close                        # roll -> flat synthetic
    assert bars[2].high / bars[2].close == 51.0 / 50.0           # range preserved
    assert all(b.symbol == "ES" for b in bars)
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/unit/test_warehouse_silver.py::test_normalize_futures_builds_adjusted_bars_with_epoch_second_ts -q`
Expected: FAIL — `ModuleNotFoundError: fxhnt.adapters.warehouse.silver_futures`.

- [ ] **Step 3: Implement**

Create `src/fxhnt/adapters/warehouse/silver_futures.py`:

```python
"""Silver: raw continuous-front futures arrays -> canonical BarRecords. Reuses the loader's
ratio-back-adjustment (continuous_front_adjust) so there is one implementation of the roll math."""
from __future__ import annotations

import numpy as np

from fxhnt.adapters.data.dbn_local import continuous_front_adjust
from fxhnt.domain.warehouse.bar import BarRecord


def normalize_futures(symbol: str, days: np.ndarray, inst: np.ndarray, close: np.ndarray,
                      high: np.ndarray, low: np.ndarray) -> list[BarRecord]:
    syn_close, syn_high, syn_low = continuous_front_adjust(inst, close, high, low)
    return [BarRecord(symbol=symbol, ts=int(d) * 86_400, close=float(c), high=float(h), low=float(lo))
            for d, c, h, lo in zip(days, syn_close, syn_high, syn_low, strict=True)]
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest tests/unit/test_warehouse_silver.py::test_normalize_futures_builds_adjusted_bars_with_epoch_second_ts -q`
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/adapters/warehouse/silver_futures.py tests/unit/test_warehouse_silver.py
git commit -m "feat(warehouse): silver_futures normalizer (reuses continuous_front_adjust)"
```

---

## Task 5: Silver normalizer — crypto

**Files:**
- Create: `src/fxhnt/adapters/warehouse/silver_crypto.py`
- Test: `tests/unit/test_warehouse_silver.py` (append)

- [ ] **Step 1: Append the failing test**

```python
def test_normalize_crypto_builds_close_only_bars() -> None:
    from fxhnt.adapters.warehouse.silver_crypto import normalize_crypto
    day = np.array([19_000, 19_001], dtype=np.int64)   # epoch-days
    close = np.array([42000.0, 43000.0])
    bars = normalize_crypto("BTCUSDT", day, close)
    assert [b.ts for b in bars] == [19_000 * 86_400, 19_001 * 86_400]
    # crypto bronze is close-only -> high == low == close (flat bar)
    assert bars[0].high == bars[0].low == bars[0].close == 42000.0
    assert all(b.symbol == "BTCUSDT" for b in bars)
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/unit/test_warehouse_silver.py::test_normalize_crypto_builds_close_only_bars -q`
Expected: FAIL — `ModuleNotFoundError: fxhnt.adapters.warehouse.silver_crypto`.

- [ ] **Step 3: Implement**

Create `src/fxhnt/adapters/warehouse/silver_crypto.py`:

```python
"""Silver: raw crypto_pit arrays (close-only daily) -> canonical BarRecords. The bronze .npz stores
only close + quote-volume, so high == low == close (a flat bar) — honest given the source."""
from __future__ import annotations

import numpy as np

from fxhnt.domain.warehouse.bar import BarRecord


def normalize_crypto(symbol: str, day: np.ndarray, close: np.ndarray) -> list[BarRecord]:
    return [BarRecord(symbol=symbol, ts=int(d) * 86_400, close=float(c), high=float(c), low=float(c))
            for d, c in zip(day, close, strict=True)]
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest tests/unit/test_warehouse_silver.py -q`
Expected: PASS (2 passed — both silver tests).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/adapters/warehouse/silver_crypto.py tests/unit/test_warehouse_silver.py
git commit -m "feat(warehouse): silver_crypto normalizer (close-only daily bars)"
```

---

## Task 6: `WarehouseIngest` service (gold write)

**Files:**
- Create: `src/fxhnt/application/warehouse_ingest.py`
- Test: `tests/integration/test_warehouse_ingest_catalog.py`

- [ ] **Step 1: Write the failing test**

```python
# tests/integration/test_warehouse_ingest_catalog.py
"""Gold ingest + catalog over a temp DuckDB warehouse."""
from __future__ import annotations

from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
from fxhnt.application.warehouse_ingest import WarehouseIngest
from fxhnt.domain.warehouse.bar import BarRecord


def test_ingest_bars_writes_feature_rows(tmp_path) -> None:
    store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    bars = [BarRecord("ES", 0, 100.0, 101.0, 99.0), BarRecord("ES", 86_400, 102.0, 103.0, 101.0)]
    n = WarehouseIngest(store).ingest_bars("ES", bars)
    assert n == 2
    rows = store.read_features("ES", features=["close", "high", "low"])
    assert [ts for ts, _ in rows] == [0, 86_400]
    assert rows[0][1] == {"close": 100.0, "high": 101.0, "low": 99.0}
    store.close()


def test_ingest_is_idempotent(tmp_path) -> None:
    store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    bars = [BarRecord("ES", 0, 100.0, 101.0, 99.0)]
    WarehouseIngest(store).ingest_bars("ES", bars)
    WarehouseIngest(store).ingest_bars("ES", bars)          # re-ingest same range
    rows = store.read_features("ES", features=["close"])
    assert len(rows) == 1                                    # no duplication
    store.close()
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/integration/test_warehouse_ingest_catalog.py -q`
Expected: FAIL — `ModuleNotFoundError: fxhnt.application.warehouse_ingest`.

- [ ] **Step 3: Implement**

Create `src/fxhnt/application/warehouse_ingest.py`:

```python
"""WarehouseIngest — the gold write step. Turns BarRecords into point-in-time feature rows and persists
them to the warehouse (idempotent on the (symbol, ts-range) being written)."""
from __future__ import annotations

from fxhnt.domain.warehouse.bar import BarRecord
from fxhnt.ports.warehouse import FeatureStore


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

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest tests/integration/test_warehouse_ingest_catalog.py -q`
Expected: PASS (2 passed).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/warehouse_ingest.py tests/integration/test_warehouse_ingest_catalog.py
git commit -m "feat(warehouse): WarehouseIngest gold-write service (idempotent)"
```

---

## Task 7: `WarehousePriceProvider` (SSOT read path)

**Files:**
- Create: `src/fxhnt/adapters/warehouse/price_provider.py`
- Test: `tests/unit/test_warehouse_price_provider.py`

- [ ] **Step 1: Write the failing test**

```python
# tests/unit/test_warehouse_price_provider.py
"""WarehousePriceProvider reads bars from the SSOT as a PriceSeries (DataProvider port)."""
from __future__ import annotations

import numpy as np

from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
from fxhnt.adapters.warehouse.price_provider import WarehousePriceProvider
from fxhnt.application.warehouse_ingest import WarehouseIngest
from fxhnt.domain.models import AssetClass, Market
from fxhnt.domain.warehouse.bar import BarRecord

_M = Market(symbol="ES", asset_class=AssetClass.FUTURE)


def _store(tmp_path):
    store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    bars = [BarRecord("ES", 1_577_836_800, 100.0, 101.0, 99.0),     # 2020-01-01
            BarRecord("ES", 1_577_923_200, 102.0, 103.0, 101.0),    # 2020-01-02
            BarRecord("ES", 1_578_009_600, 104.0, 105.0, 103.0)]    # 2020-01-03
    WarehouseIngest(store).ingest_bars("ES", bars)
    return store


def test_fetch_returns_priceseries_round_trip(tmp_path) -> None:
    ps = WarehousePriceProvider(_store(tmp_path)).fetch(_M)
    assert list(ps.dates) == ["2020-01-01", "2020-01-02", "2020-01-03"]
    assert list(ps.close) == [100.0, 102.0, 104.0]
    assert list(ps.high) == [101.0, 103.0, 105.0]
    assert list(ps.low) == [99.0, 101.0, 103.0]
    assert ps.market == _M


def test_fetch_respects_start_end_no_lookahead(tmp_path) -> None:
    ps = WarehousePriceProvider(_store(tmp_path)).fetch(_M, start="2020-01-02", end="2020-01-02")
    assert list(ps.dates) == ["2020-01-02"]            # bars outside [start,end] excluded
    assert list(ps.close) == [102.0]


def test_fetch_raises_when_symbol_absent(tmp_path) -> None:
    import pytest
    with pytest.raises(RuntimeError):
        WarehousePriceProvider(_store(tmp_path)).fetch(Market(symbol="NQ", asset_class=AssetClass.FUTURE))
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/unit/test_warehouse_price_provider.py -q`
Expected: FAIL — `ModuleNotFoundError: fxhnt.adapters.warehouse.price_provider`.

- [ ] **Step 3: Implement**

Create `src/fxhnt/adapters/warehouse/price_provider.py`:

```python
"""WarehousePriceProvider — implements the DataProvider port by reading OHLC bars from the warehouse
SSOT and returning a PriceSeries. Consumers (research/forward) read the SSOT with no change to the
backtest. Reads are bounded by start/end (and the warehouse is point-in-time by construction)."""
from __future__ import annotations

import datetime as dt

import numpy as np

from fxhnt.domain.models import Market, PriceSeries
from fxhnt.ports.warehouse import FeatureStore


def _iso_to_ts(iso: str | None) -> int | None:
    if iso is None:
        return None
    return int(dt.datetime.strptime(iso, "%Y-%m-%d").replace(tzinfo=dt.timezone.utc).timestamp())


def _ts_to_iso(ts: int) -> str:
    return dt.datetime.fromtimestamp(int(ts), dt.timezone.utc).strftime("%Y-%m-%d")


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=["close", "high", "low"])
        if not rows:
            raise RuntimeError(f"warehouse has no bars for {market.symbol}")
        dates = tuple(_ts_to_iso(ts) for ts, _ in rows)
        close = np.array([f["close"] for _, f in rows], dtype=float)
        high = np.array([f.get("high", f["close"]) for _, f in rows], dtype=float)
        low = np.array([f.get("low", f["close"]) for _, f in rows], dtype=float)
        return PriceSeries(market=market, dates=dates, close=close, high=high, low=low)
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest tests/unit/test_warehouse_price_provider.py -q`
Expected: PASS (3 passed).

- [ ] **Step 5: Verify mypy**

Run: `python3 -m mypy src/fxhnt/adapters/warehouse/price_provider.py`
Expected: `Success: no issues found`.

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/adapters/warehouse/price_provider.py tests/unit/test_warehouse_price_provider.py
git commit -m "feat(warehouse): WarehousePriceProvider reads bars as PriceSeries (DataProvider)"
```

---

## Task 8: Catalog / freshness

**Files:**
- Modify: `src/fxhnt/adapters/warehouse/duckdb_feature_store.py` (add `catalog`)
- Create: `src/fxhnt/application/warehouse_catalog.py`
- Test: `tests/integration/test_warehouse_ingest_catalog.py` (append) + `tests/unit/test_warehouse_catalog_format.py`

- [ ] **Step 1: Append the failing catalog test (store-level)**

Append to `tests/integration/test_warehouse_ingest_catalog.py`:

```python
def test_catalog_reports_count_and_range(tmp_path) -> None:
    store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    WarehouseIngest(store).ingest_bars("ES", [BarRecord("ES", 0, 100.0, 101.0, 99.0),
                                              BarRecord("ES", 86_400, 102.0, 103.0, 101.0)])
    WarehouseIngest(store).ingest_bars("NQ", [BarRecord("NQ", 86_400, 200.0, 201.0, 199.0)])
    cat = dict((sym, (n, lo, hi)) for sym, n, lo, hi in store.catalog())
    assert cat["ES"] == (2, 0, 86_400)
    assert cat["NQ"] == (1, 86_400, 86_400)
    store.close()
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest "tests/integration/test_warehouse_ingest_catalog.py::test_catalog_reports_count_and_range" -q`
Expected: FAIL — `DuckDbFeatureStore` has no `catalog`.

- [ ] **Step 3: Implement `catalog` on the store**

In `src/fxhnt/adapters/warehouse/duckdb_feature_store.py`, add a method (before `close`):

```python
    def catalog(self) -> list[tuple[str, int, int, int]]:
        """Per symbol: (symbol, bar_count, first_ts, last_ts) over the 'close' feature — the data
        inventory + freshness view of the SSOT."""
        return self._con.execute(
            "SELECT symbol, count(*), min(ts), max(ts) FROM features WHERE feature = 'close' "
            "GROUP BY symbol ORDER BY symbol"
        ).fetchall()
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest "tests/integration/test_warehouse_ingest_catalog.py::test_catalog_reports_count_and_range" -q`
Expected: PASS.

- [ ] **Step 5: Write the failing format test**

```python
# tests/unit/test_warehouse_catalog_format.py
"""The catalog formatter renders the SSOT inventory as a readable table."""
from __future__ import annotations

from fxhnt.application.warehouse_catalog import format_catalog


def test_format_catalog_renders_symbols_and_dates() -> None:
    out = format_catalog([("ES", 2, 0, 86_400), ("NQ", 1, 86_400, 86_400)])
    assert "ES" in out and "NQ" in out
    assert "1970-01-01" in out and "1970-01-02" in out   # ts 0 and 86400 as ISO days
    assert "2" in out                                     # ES bar count
```

- [ ] **Step 6: Run test to verify it fails**

Run: `python3 -m pytest tests/unit/test_warehouse_catalog_format.py -q`
Expected: FAIL — `ModuleNotFoundError: fxhnt.application.warehouse_catalog`.

- [ ] **Step 7: Implement the formatter**

Create `src/fxhnt/application/warehouse_catalog.py`:

```python
"""Catalog/freshness view — render the warehouse SSOT inventory (per-symbol bar count + date range)."""
from __future__ import annotations

import datetime as dt


def _iso(ts: int) -> str:
    return dt.datetime.fromtimestamp(int(ts), dt.timezone.utc).strftime("%Y-%m-%d")


def format_catalog(rows: list[tuple[str, int, int, int]]) -> str:
    if not rows:
        return "warehouse is empty"
    header = f"{'symbol':<12}{'bars':>8}  {'first':<12}{'last':<12}"
    lines = [header, "-" * len(header)]
    for symbol, n, first_ts, last_ts in rows:
        lines.append(f"{symbol:<12}{n:>8}  {_iso(first_ts):<12}{_iso(last_ts):<12}")
    return "\n".join(lines)
```

- [ ] **Step 8: Run tests to verify they pass**

Run: `python3 -m pytest tests/unit/test_warehouse_catalog_format.py tests/integration/test_warehouse_ingest_catalog.py -q`
Expected: PASS (all).

- [ ] **Step 9: Commit**

```bash
git add src/fxhnt/adapters/warehouse/duckdb_feature_store.py src/fxhnt/application/warehouse_catalog.py \
        tests/unit/test_warehouse_catalog_format.py tests/integration/test_warehouse_ingest_catalog.py
git commit -m "feat(warehouse): catalog/freshness inventory (store.catalog + format_catalog)"
```

---

## Task 9: CLI wiring — `warehouse-ingest`, `warehouse-catalog`, and `warehouse` data source

**Files:**
- Modify: `src/fxhnt/cli.py` (add 2 commands; extend `_data_provider`)
- Test: `tests/unit/test_data_provider_warehouse.py`

- [ ] **Step 1: Write the failing test (the cutover seam)**

```python
# tests/unit/test_data_provider_warehouse.py
"""The data_source seam selects the warehouse SSOT provider."""
from __future__ import annotations

from fxhnt.cli import _data_provider
from fxhnt.config import Settings


def test_data_provider_warehouse_returns_warehouse_provider(tmp_path) -> None:
    from fxhnt.adapters.warehouse.price_provider import WarehousePriceProvider
    s = Settings(warehouse_path=str(tmp_path / "wh.duckdb"))
    provider = _data_provider("warehouse", s)
    assert isinstance(provider, WarehousePriceProvider)
    assert provider.name == "warehouse"
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/unit/test_data_provider_warehouse.py -q`
Expected: FAIL — `_data_provider("warehouse", ...)` does not return a `WarehousePriceProvider` (likely raises/returns the default).

- [ ] **Step 3: Extend `_data_provider`**

Find `_data_provider` in `src/fxhnt/cli.py` (it currently maps `yahoo`/`databento`). Add a `warehouse` branch. Read the existing function first; the added branch is:

```python
    if data_source == "warehouse":
        from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
        from fxhnt.adapters.warehouse.price_provider import WarehousePriceProvider
        return WarehousePriceProvider(DuckDbFeatureStore(settings.warehouse_path))
```

(Place it before the existing yahoo/databento handling so the existing behavior is otherwise unchanged.)

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest tests/unit/test_data_provider_warehouse.py -q`
Expected: PASS.

- [ ] **Step 5: Add the `warehouse-ingest` command**

Add to `src/fxhnt/cli.py` (near the other `@app.command()` data commands, e.g. after `fetch`):

```python
@app.command("warehouse-ingest")
def warehouse_ingest(
    data_dir: str = typer.Option(..., "--data-dir", help="dir with crypto_pit/ and/or <symbol>.dbn bronze data"),
    source: str = typer.Option(..., "--source", help="futures | crypto"),
    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.duckdb_feature_store import DuckDbFeatureStore
    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

    settings = get_settings()
    settings.ensure_dirs()
    store = DuckDbFeatureStore(settings.warehouse_path)
    ingest = WarehouseIngest(store)

    if source == "futures":
        if not symbol:
            typer.echo("error: --symbol is required for --source futures", err=True)
            raise typer.Exit(1)
        from fxhnt.adapters.data.dbn_local import read_front_arrays
        days, inst, close, high, low = read_front_arrays(os.path.join(data_dir, f"{symbol}.dbn"), symbol)
        n = ingest.ingest_bars(symbol, normalize_futures(symbol, days, inst, close, high, low))
        typer.echo(f"futures {symbol}: {n} bars -> warehouse (normalized via silver_futures.normalize_futures)")
    elif source == "crypto":
        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
                bars = normalize_crypto(sym, z["day"], z["close"])
            total += ingest.ingest_bars(sym, bars)
        typer.echo(f"crypto: {total} bars -> warehouse")
    else:
        typer.echo("error: --source must be 'futures' or 'crypto'", err=True)
        raise typer.Exit(1)
    store.close()
```

(Note: the stray `from ... import normalize_futures as _nf` line above is a mistake — do NOT include it; `normalize_futures` is imported from `silver_futures` at the top of the command. Import only `read_front_arrays` from `dbn_local`.)

- [ ] **Step 6: Add the `warehouse-catalog` command**

```python
@app.command("warehouse-catalog")
def warehouse_catalog() -> None:
    """Show the warehouse SSOT inventory: per-symbol bar count + date range (freshness)."""
    from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
    from fxhnt.application.warehouse_catalog import format_catalog

    settings = get_settings()
    store = DuckDbFeatureStore(settings.warehouse_path)
    typer.echo(format_catalog(store.catalog()))
    store.close()
```

- [ ] **Step 7: Verify CLI imports + full suite + mypy**

Run: `python3 -c "import fxhnt.cli"` (catches import errors), then `python3 -m pytest tests/ -q`, then `python3 -m mypy src/fxhnt/cli.py`
Expected: clean import, all tests pass, mypy `Success`.

- [ ] **Step 8: Manual smoke (no network) — catalog on an empty warehouse**

Run: `python3 -m fxhnt warehouse-catalog` (or `fxhnt warehouse-catalog` if installed)
Expected: prints `warehouse is empty` (fresh DB) — confirms the command wires end-to-end.

- [ ] **Step 9: Commit**

```bash
git add src/fxhnt/cli.py tests/unit/test_data_provider_warehouse.py
git commit -m "feat(cli): warehouse-ingest + warehouse-catalog commands; warehouse data source"
```

---

## Final verification

- [ ] **Full suite + mypy on all new/changed files**

Run:
```bash
python3 -m pytest tests/ -q
python3 -m mypy src/fxhnt/domain/warehouse src/fxhnt/adapters/warehouse src/fxhnt/application/warehouse_ingest.py src/fxhnt/application/warehouse_catalog.py src/fxhnt/adapters/data/dbn_local.py src/fxhnt/cli.py src/fxhnt/config.py
```
Expected: all tests pass; mypy `Success`.

- [ ] **SSOT equivalence sanity (the point of the increment)**

The round-trip test (`test_fetch_returns_priceseries_round_trip`) already proves a `PriceSeries` built from the warehouse equals the bars that were ingested. With real data, `fxhnt warehouse-ingest --data-dir <dir> --source futures --symbol ES` followed by selecting `--data-source warehouse` in `hunt`/`run` makes research read the SSOT — reversible by switching the flag back to `databento`.
