# fxhnt Edge-Pipeline Containerization Implementation Plan

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

**Goal:** Port the two data fetchers into fxhnt so the combined-book (edge) daily loop runs fully in-cluster: fetch → forward-track → ingest → dashboard, laptop-independent.

**Architecture:** Two new fetch adapters in `fxhnt/adapters/data/` with the network call injected (default real, fakes in tests), a `fxhnt fetch` CLI command, and an updated cockpit CronJob that runs the full loop. The combined-book tracker is already fxhnt-native; only the fetchers were missing.

**Tech Stack:** Python 3.11, numpy, stdlib `urllib` (crypto, no key), databento (futures, key from env), typer, pytest. Repo `/home/jgrusewski/Work/fxhnt`, branch `feat/cockpit-dashboard`.

---

## File Structure

**Create:**
- `src/fxhnt/adapters/data/crypto_fetch.py` — `fetch_crypto_pit(...)` (Binance public, refresh, injected GET).
- `src/fxhnt/adapters/data/futures_fetch.py` — `fetch_futures(...)` (Databento, cost-gated, injected client).
- `tests/unit/test_crypto_fetch.py`, `tests/unit/test_futures_fetch.py`, `tests/integration/test_fetch_cli.py`

**Modify:**
- `src/fxhnt/cli.py` — add `fetch` command.
- `infra/k8s/jobs/fxhnt-forward-cronjob.yaml` — fetch → track → ingest full loop.

All commits: author `git -c user.email=jeroen@grusewski.nl -c user.name=jgrusewski`; END every commit body with `Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>`. Do NOT push. Always `cd /home/jgrusewski/Work/fxhnt` (cwd may reset).

---

# Milestone 4 — Edge fetch port

### Task 1: Crypto fetch adapter (Binance public, refresh, offline-testable)

**Files:**
- Create: `src/fxhnt/adapters/data/crypto_fetch.py`
- Test: `tests/unit/test_crypto_fetch.py`

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

Create `tests/unit/test_crypto_fetch.py`:

```python
"""Crypto fetch: offline (injected GET) — universe selection, npz keys/shape, <200-bar skip, refresh."""
from __future__ import annotations

import numpy as np

from fxhnt.adapters.data.crypto_fetch import fetch_crypto_pit


def _fake_get_factory(n_bars: int):
    """Return a `get(url)` that serves canned Binance fapi JSON for any symbol."""
    def get(url: str):
        if "exchangeInfo" in url:
            return {"symbols": [
                {"symbol": "AAAUSDT", "contractType": "PERPETUAL", "quoteAsset": "USDT", "status": "TRADING"},
                {"symbol": "BBBUSDT", "contractType": "PERPETUAL", "quoteAsset": "USDT", "status": "TRADING"},
            ]}
        if "ticker/24hr" in url:
            return [{"symbol": "AAAUSDT", "quoteVolume": "9999"},
                    {"symbol": "BBBUSDT", "quoteVolume": "10"}]
        if "klines" in url:
            # n_bars daily klines: [openTime, o,h,l, close, vol, closeTime, quoteVol, ...]
            day_ms = 86_400_000
            return [[i * day_ms, "1", "1", "1", str(100 + i), "1", i * day_ms + 1, "1000", 0, 0, 0, 0]
                    for i in range(n_bars)]
        if "fundingRate" in url:
            return []
        return None
    return get


def test_writes_npz_with_day_and_close(tmp_path) -> None:
    n, total = fetch_crypto_pit(str(tmp_path), top_n=2, get=_fake_get_factory(300), sleep=lambda s: None, dead=[])
    assert n == 2 and total == 2
    z = np.load(tmp_path / "crypto_pit" / "AAAUSDT.npz")
    assert "day" in z and "close" in z
    assert len(z["day"]) == 300 and z["close"][0] == 100.0


def test_skips_symbols_with_too_few_bars(tmp_path) -> None:
    n, total = fetch_crypto_pit(str(tmp_path), top_n=2, get=_fake_get_factory(50), sleep=lambda s: None, dead=[])
    assert n == 0 and total == 2          # <200 bars -> skipped


def test_refresh_overwrites_existing(tmp_path) -> None:
    fetch_crypto_pit(str(tmp_path), top_n=1, get=_fake_get_factory(300), sleep=lambda s: None, dead=[])
    # second run with longer series must overwrite (refresh), not skip
    fetch_crypto_pit(str(tmp_path), top_n=1, get=_fake_get_factory(400), sleep=lambda s: None, dead=[])
    z = np.load(tmp_path / "crypto_pit" / "AAAUSDT.npz")
    assert len(z["day"]) == 400
```

- [ ] **Step 2: Run tests to verify they fail**

Run: `cd /home/jgrusewski/Work/fxhnt && pytest tests/unit/test_crypto_fetch.py -v`
Expected: FAIL with `ModuleNotFoundError: No module named 'fxhnt.adapters.data.crypto_fetch'`

- [ ] **Step 3: Write minimal implementation**

Create `src/fxhnt/adapters/data/crypto_fetch.py`:

```python
"""Point-in-time crypto fetch for the combined-book momentum sleeve — Binance USDT-perp daily closes,
survivorship-free (live top-N by 24h quote-volume UNION known-dead perps). No API key (public fapi).

Ported from foxhunt scripts/surfer/fetch_crypto_pit.py with two changes for daily forward use:
  * REFRESH (overwrite) instead of skip-existing — the forward tracker needs fresh days every run;
  * a SINGLE klines page (limit=1500) instead of deep endTime pagination — ~1500 daily bars dwarfs the
    30-day momentum lookback, and one request per symbol keeps a daily refresh cheap.
The HTTP GET is injected so tests run offline."""
from __future__ import annotations

import json
import logging
import os
import time
import urllib.request
from collections.abc import Callable
from typing import Any

import numpy as np

_log = logging.getLogger(__name__)
DAY_MS = 86_400_000
# famous delisted/dead Binance USDT perps — klines remain available post-delist (the survivorship signal)
DEAD = ["LUNAUSDT", "ANCUSDT", "SRMUSDT", "HNTUSDT", "MATICUSDT", "FTTUSDT", "RAYUSDT", "WAVESUSDT",
        "BNXUSDT", "SCUSDT", "OCEANUSDT", "AGIXUSDT", "CVCUSDT", "TLMUSDT", "DENTUSDT", "KEYUSDT",
        "CTKUSDT", "TOMOUSDT", "AKROUSDT", "BLZUSDT", "COMBOUSDT", "FTMUSDT", "DGBUSDT", "REEFUSDT",
        "SLPUSDT", "IDEXUSDT", "LINAUSDT", "NEBLUSDT", "RADUSDT", "BTSUSDT", "STMXUSDT", "MDTUSDT",
        "AMBUSDT", "GTCUSDT", "DARUSDT"]


def _default_get(url: str) -> Any:
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "curl/8"})
        return json.load(urllib.request.urlopen(req, timeout=30))
    except Exception as e:               # noqa: BLE001 — network best-effort; caller logs/skips
        _log.warning("crypto GET failed (%s): %s", url, e)
        return None


def _universe(n: int, get: Callable[[str], Any]) -> list[str]:
    info = get("https://fapi.binance.com/fapi/v1/exchangeInfo")
    perps = {s["symbol"] for s in info["symbols"]
             if s.get("contractType") == "PERPETUAL" and s.get("quoteAsset") == "USDT"
             and s.get("status") == "TRADING"}
    tick = get("https://fapi.binance.com/fapi/v1/ticker/24hr")
    vol = {t["symbol"]: float(t["quoteVolume"]) for t in tick if t["symbol"] in perps}
    return sorted(vol, key=lambda s: -vol[s])[:n]


def _klines(sym: str, get: Callable[[str], Any]) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    k = get(f"https://fapi.binance.com/fapi/v1/klines?symbol={sym}&interval=1d&limit=1500")
    if not k:
        return np.array([]), np.array([]), np.array([])
    d = {int(r[0]): (float(r[4]), float(r[7])) for r in k}   # close, quote-volume
    days = np.array(sorted(d))
    cl = np.array([d[t][0] for t in days])
    qv = np.array([d[t][1] for t in days])
    return days // DAY_MS, cl, qv


def _funding_daily(sym: str, get: Callable[[str], Any]) -> dict[int, float]:
    f = get(f"https://fapi.binance.com/fapi/v1/fundingRate?symbol={sym}&limit=1000")
    daily: dict[int, float] = {}
    for r in f or []:
        daily[int(r["fundingTime"]) // DAY_MS] = daily.get(int(r["fundingTime"]) // DAY_MS, 0.0) + float(r["fundingRate"])
    return daily


def fetch_crypto_pit(data_dir: str, *, top_n: int = 120, get: Callable[[str], Any] = _default_get,
                     sleep: Callable[[float], None] = time.sleep, dead: list[str] | None = None) -> tuple[int, int]:
    """Fetch the survivorship-free crypto panel into <data_dir>/crypto_pit/<SYM>.npz (keys: day, close, qvol,
    funding). REFRESH semantics (overwrites). Returns (symbols_written, symbols_attempted)."""
    out_dir = os.path.join(data_dir, "crypto_pit")
    os.makedirs(out_dir, exist_ok=True)
    syms = sorted(set(_universe(top_n, get)) | set(DEAD if dead is None else dead))
    ok = 0
    for sym in syms:
        kd, cl, qv = _klines(sym, get)
        if len(kd) < 200:
            continue
        fund = _funding_daily(sym, get)
        fdays = np.array(sorted(fund)) if fund else np.array([])
        frate = np.array([fund[d] for d in sorted(fund)]) if fund else np.array([])
        np.savez(os.path.join(out_dir, f"{sym}.npz"), day=kd, close=cl, qvol=qv, funding=frate, funding_day=fdays)
        ok += 1
        sleep(0.1)
    _log.info("crypto fetch: %d/%d symbols -> %s", ok, len(syms), out_dir)
    return ok, len(syms)
```

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

Run: `cd /home/jgrusewski/Work/fxhnt && pytest tests/unit/test_crypto_fetch.py -v`
Expected: PASS (3 passed)

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/adapters/data/crypto_fetch.py tests/unit/test_crypto_fetch.py
git commit -m "feat(fetch): port crypto_pit fetcher into fxhnt (Binance public, refresh, offline-testable)"
```

---

### Task 2: Futures fetch adapter (Databento, cost-gated, offline-testable)

**Files:**
- Create: `src/fxhnt/adapters/data/futures_fetch.py`
- Test: `tests/unit/test_futures_fetch.py`

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

Create `tests/unit/test_futures_fetch.py`:

```python
"""Futures fetch: injected fake Databento client — per-root cost gate, cumulative cap, write path. No network."""
from __future__ import annotations

from fxhnt.adapters.data.futures_fetch import fetch_futures


class _FakeData:
    def __init__(self, n): self._n = n
    def to_file(self, path): open(path, "w").close()
    def __iter__(self): return iter(range(self._n))


class _FakeClient:
    """Mimics databento.Historical: .metadata.get_cost / .get_dataset_range, .timeseries.get_range."""
    def __init__(self, cost): self._cost = cost; self.metadata = self; self.timeseries = self
    def get_dataset_range(self, dataset): return {"end": "2026-06-13"}
    def get_cost(self, **kw): return self._cost
    def get_range(self, **kw): return _FakeData(100)


def test_fetches_within_caps(tmp_path) -> None:
    recs, ok, spent = fetch_futures(str(tmp_path), roots=["ES", "NQ"], start="2020-01-01", end="2026-06-30",
                                    max_cost_usd=40.0, per_root_cap_usd=10.0, client=_FakeClient(0.30))
    assert ok == 2 and recs == 200
    assert abs(spent - 0.60) < 1e-9
    assert (tmp_path / "ES.dbn").exists() and (tmp_path / "NQ.dbn").exists()


def test_per_root_cap_skips_expensive_root(tmp_path) -> None:
    recs, ok, spent = fetch_futures(str(tmp_path), roots=["ES"], start="2020-01-01", end="2026-06-30",
                                    max_cost_usd=40.0, per_root_cap_usd=0.10, client=_FakeClient(0.30))
    assert ok == 0 and recs == 0 and spent == 0.0       # 0.30 > 0.10 per-root cap -> skipped, no download
    assert not (tmp_path / "ES.dbn").exists()


def test_cumulative_cap_stops_further_roots(tmp_path) -> None:
    recs, ok, spent = fetch_futures(str(tmp_path), roots=["ES", "NQ", "YM"], start="2020-01-01",
                                    end="2026-06-30", max_cost_usd=0.50, per_root_cap_usd=10.0,
                                    client=_FakeClient(0.30))
    assert ok == 1 and abs(spent - 0.30) < 1e-9         # ES ok (0.30); NQ would breach 0.50 -> stop
```

- [ ] **Step 2: Run tests to verify they fail**

Run: `cd /home/jgrusewski/Work/fxhnt && pytest tests/unit/test_futures_fetch.py -v`
Expected: FAIL with `ModuleNotFoundError: No module named 'fxhnt.adapters.data.futures_fetch'`

- [ ] **Step 3: Write minimal implementation**

Create `src/fxhnt/adapters/data/futures_fetch.py`:

```python
"""Continuous front-month-input futures fetch for the trend hedge sleeve — Databento GLBX ohlcv-1d, parent
symbology, written as <data_dir>/<ROOT>.dbn (LocalDbnLoader builds the continuous series from these).

Ported from foxhunt scripts/surfer/fetch_daily.py. Preserves the hard-won cost discipline: estimate
get_cost (free, no download) and refuse any root over the per-root cap or any fetch that would breach the
cumulative cap BEFORE downloading. The Databento client is injected so tests exercise the gates with no
network and no spend. A runaway download once cost $3.5k — the gate is non-negotiable."""
from __future__ import annotations

import logging
import os
import time
from typing import Any

_log = logging.getLogger(__name__)

DATASET = "GLBX.MDP3"
SCHEMA = "ohlcv-1d"
STYPE = "parent"
DEFAULT_ROOTS = ["ES", "NQ", "YM", "RTY", "ZN", "ZB", "ZF", "ZT", "6E", "6J", "6B", "6A",
                 "6C", "GC", "SI", "HG", "CL", "NG", "RB", "ZC", "ZS", "ZW"]


def _default_client() -> Any:
    import databento as db
    key = os.environ.get("DATABENTO_API_KEY")
    if not key:
        raise RuntimeError("DATABENTO_API_KEY not set (required for --futures)")
    return db.Historical(key)


def fetch_futures(data_dir: str, *, roots: list[str] | None = None, start: str = "2010-06-06",
                  end: str = "", max_cost_usd: float = 40.0, per_root_cap_usd: float = 10.0,
                  client: Any | None = None) -> tuple[int, int, float]:
    """Fetch per-root ohlcv-1d DBN into <data_dir>/<ROOT>.dbn under cost caps. Returns (records, roots_ok,
    spent_usd). `end` defaults to today clamped to the dataset's available range."""
    import datetime as dt

    roots = roots or DEFAULT_ROOTS
    client = client or _default_client()
    end = end or dt.date.today().isoformat()
    try:                                                   # clamp END to dataset range (GLBX lags 1-3d)
        avail = (client.metadata.get_dataset_range(dataset=DATASET).get("end") or "")[:10]
        if avail:
            end = min(end, avail)
    except Exception as e:                                 # noqa: BLE001 — clamp is best-effort
        _log.warning("dataset range clamp failed: %s", e)
    os.makedirs(data_dir, exist_ok=True)

    total_recs, spent, ok = 0, 0.0, 0
    for r in roots:
        sym = f"{r}.FUT"
        try:
            c_r = client.metadata.get_cost(dataset=DATASET, symbols=[sym], schema=SCHEMA,
                                           start=start, end=end, stype_in=STYPE)
        except Exception as e:                             # noqa: BLE001
            _log.warning("%s: get_cost failed (%s); skip", r, e)
            continue
        if c_r > per_root_cap_usd or spent + c_r > max_cost_usd:
            _log.warning("%s: cost $%.4f would breach cap (spent $%.2f); skip", r, c_r, spent)
            continue
        for attempt in range(1, 4):
            try:
                data = client.timeseries.get_range(dataset=DATASET, symbols=[sym], schema=SCHEMA,
                                                   start=start, end=end, stype_in=STYPE)
                data.to_file(os.path.join(data_dir, f"{r}.dbn"))
                total_recs += sum(1 for _ in data)
                spent += c_r
                ok += 1
                break
            except Exception as e:                         # noqa: BLE001
                _log.warning("%s: attempt %d failed (%s)", r, attempt, e)
                time.sleep(3)
    _log.info("futures fetch: %d/%d roots, %d records, $%.2f", ok, len(roots), total_recs, spent)
    return total_recs, ok, spent
```

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

Run: `cd /home/jgrusewski/Work/fxhnt && pytest tests/unit/test_futures_fetch.py -v`
Expected: PASS (3 passed)

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/adapters/data/futures_fetch.py tests/unit/test_futures_fetch.py
git commit -m "feat(fetch): port futures fetcher into fxhnt (Databento, cost-gated, offline-testable)"
```

---

### Task 3: `fxhnt fetch` CLI command

**Files:**
- Modify: `src/fxhnt/cli.py`
- Test: `tests/integration/test_fetch_cli.py`

- [ ] **Step 1: Read the CLI head to confirm patterns**

Run: `sed -n '1,25p' src/fxhnt/cli.py`
Expected: confirm the `app = typer.Typer(...)` object and `get_settings`/`Settings` import. The CLI uses **typer**. Adapt the code below to match the real `app` name and settings accessor.

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

Create `tests/integration/test_fetch_cli.py`:

```python
"""`fxhnt fetch` wiring: runs the selected fetchers into the data dir (fetchers monkeypatched — no network)."""
from __future__ import annotations

from typer.testing import CliRunner

from fxhnt import cli

runner = CliRunner()


def test_fetch_crypto_only_invokes_crypto(monkeypatch, tmp_path) -> None:
    calls = {}
    monkeypatch.setattr(cli, "fetch_crypto_pit", lambda data_dir, **kw: calls.setdefault("crypto", data_dir) or (5, 5))
    monkeypatch.setattr(cli, "fetch_futures", lambda data_dir, **kw: calls.setdefault("futures", data_dir) or (0, 0, 0.0))
    res = runner.invoke(cli.app, ["fetch", "--data-dir", str(tmp_path), "--crypto", "--no-futures"])
    assert res.exit_code == 0
    assert calls.get("crypto") == str(tmp_path) and "futures" not in calls


def test_fetch_futures_invokes_futures(monkeypatch, tmp_path) -> None:
    calls = {}
    monkeypatch.setattr(cli, "fetch_crypto_pit", lambda data_dir, **kw: (0, 0))
    monkeypatch.setattr(cli, "fetch_futures", lambda data_dir, **kw: calls.setdefault("futures", data_dir) or (10, 1, 0.3))
    res = runner.invoke(cli.app, ["fetch", "--data-dir", str(tmp_path), "--no-crypto", "--futures"])
    assert res.exit_code == 0 and calls.get("futures") == str(tmp_path)
```

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

Run: `cd /home/jgrusewski/Work/fxhnt && pytest tests/integration/test_fetch_cli.py -v`
Expected: FAIL (no `fetch` command; `cli.fetch_crypto_pit` attribute missing for monkeypatch).

- [ ] **Step 4: Add the command**

At the TOP of `src/fxhnt/cli.py` add module-level imports so the test can monkeypatch them on `cli`:

```python
from fxhnt.adapters.data.crypto_fetch import fetch_crypto_pit
from fxhnt.adapters.data.futures_fetch import fetch_futures
```

Then append the command (adapt `get_settings()` to the file's real accessor):

```python
@app.command()
def fetch(
    data_dir: str = typer.Option(..., "--data-dir", help="dir for crypto_pit/ and <root>.dbn"),
    crypto: bool = typer.Option(True, "--crypto/--no-crypto"),
    futures: bool = typer.Option(True, "--futures/--no-futures"),
) -> None:
    """Fetch the combined-book data into data_dir: crypto (Binance public, free) and/or futures (Databento)."""
    if crypto:
        n, total = fetch_crypto_pit(data_dir)
        typer.echo(f"crypto: {n}/{total} symbols -> {data_dir}/crypto_pit")
    if futures:
        s = get_settings().databento
        recs, ok, spent = fetch_futures(data_dir, start=s.default_start, max_cost_usd=s.max_cost_usd)
        typer.echo(f"futures: {ok} roots, {recs} records, ${spent:.2f} -> {data_dir}")
```

Note: `Settings.databento` has `default_start` and `max_cost_usd` (confirm names in `src/fxhnt/config.py`; `max_cost_usd` defaults to 0.50 there — for the real all-roots fetch the CronJob passes a higher cap via env `FXHNT_DATABENTO_MAX_COST_USD`, or raise the default; do NOT hardcode a high cap in code). If `default_start` is `2018-05-01`, that is fine for the hedge sleeve.

- [ ] **Step 5: Run test + full suite**

Run: `cd /home/jgrusewski/Work/fxhnt && pytest tests/integration/test_fetch_cli.py -v && pytest -q`
Expected: fetch CLI tests PASS; whole suite green.

- [ ] **Step 6: Lint**

Run: `ruff check src/fxhnt/adapters/data/crypto_fetch.py src/fxhnt/adapters/data/futures_fetch.py`
Expected: clean (fix any issues inline, re-run).

- [ ] **Step 7: Commit**

```bash
git add src/fxhnt/cli.py tests/integration/test_fetch_cli.py
git commit -m "feat(fetch): fxhnt fetch CLI (crypto/futures into data-dir)"
```

---

# Milestone 5 — CronJob full loop

### Task 4: Update the cockpit CronJob to fetch → track → ingest

**Files:**
- Modify: `infra/k8s/jobs/fxhnt-forward-cronjob.yaml`

- [ ] **Step 1: Replace the CronJob command/args with the full loop**

In `infra/k8s/jobs/fxhnt-forward-cronjob.yaml`, change the container `command`/`args` so the daily job runs
fetch → forward-track → ingest. Crypto runs daily (free); futures only on Sundays (cost control). Use a
`sh -c` script:

```yaml
              command: ["sh", "-c"]
              args:
                - |
                  set -e
                  fxhnt migrate-cockpit
                  fxhnt fetch --data-dir /data/surfer --crypto --no-futures
                  if [ "$(date -u +%u)" = "7" ]; then
                    fxhnt fetch --data-dir /data/surfer --no-crypto --futures
                  fi
                  fxhnt forward-track --data-dir /data/surfer --state /data/surfer/fxhnt_combined_forward.json
                  fxhnt ingest-forward --state-dir /data/surfer
```

Keep the existing `env` (the `FXHNT_OPERATIONAL_DSN` from `COCKPIT_DSN` and `DATABENTO_API_KEY` from
`foxhunt-secrets`), the `volumeMounts` (`/data`), the PVC, `nodeSelector`, and `imagePullSecrets`. Also add an
env var for the futures cost cap so it is not hardcoded:

```yaml
                - name: FXHNT_DATABENTO_MAX_COST_USD
                  value: "40.0"
```

- [ ] **Step 2: Validate (server dry-run, non-mutating)**

Run: `cd /home/jgrusewski/Work/fxhnt && kubectl apply --dry-run=server -f infra/k8s/jobs/fxhnt-forward-cronjob.yaml`
Expected: `persistentvolumeclaim/fxhnt-surfer-data ... (server dry run)` and `cronjob.batch/fxhnt-forward ... (server dry run)` with no schema errors. If the cluster is unreachable, fall back to `--dry-run=client` and note it.

- [ ] **Step 3: Commit**

```bash
git add infra/k8s/jobs/fxhnt-forward-cronjob.yaml
git commit -m "deploy(cockpit): CronJob full loop — fetch -> forward-track -> ingest (futures weekly)"
```

---

## Self-Review notes

- **Spec coverage:** crypto_fetch (T1) · futures_fetch (T2) · `fxhnt fetch` CLI (T3) · CronJob full loop (T4).
  Edge-only scope — funding-harvest and other trackers explicitly deferred. Live deploy remains the cockpit
  plan's user-gated Task 17.
- **No network in tests:** both fetchers inject their I/O (GET callable / Databento client); CLI test
  monkeypatches the fetchers. The cost gate is exercised without spend.
- **Type consistency:** `fetch_crypto_pit(data_dir, *, top_n, get, sleep, dead) -> (int, int)` and
  `fetch_futures(data_dir, *, roots, start, end, max_cost_usd, per_root_cap_usd, client) -> (int, int, float)`
  used identically in the CLI and tests. npz keys (`day`, `close`, `qvol`, `funding`) match `load_crypto_panel`'s
  required `day`+`close`. `<root>.dbn` path matches `LocalDbnLoader.load`.
- **No placeholders.** Behavioral change (crypto refresh + single-page) is explicit and tested.
