# Lifecycle B1 — Paper-Track Strategies as Clean Dagster Assets — 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:** Port the 6 remaining paper-forward strategies (`funding`, `crossvenue`, `multistrat`, `sixtyforty`, `gd`, `poc`) into clean hexagonal fxhnt services + Dagster assets that each write a `days-list` `*_state.json` the existing `cockpit_forward` asset ingests, so all 7 registry strategies show live forward NAV; then retire the foxhunt `scripts/surfer/*` crons.

**Architecture:** A single generic `ForwardTracker` drives every strategy through one `ForwardStrategy.advance()` port (two patterns: recomputable-series and live-booking-with-positions). Each strategy = a pure-numpy domain module + an application service composing a market-data port; one Dagster `@asset` per strategy produces its state file. No torch, no warehouse/PIT deps, no `state_reader` change (all emit `days-list`).

**Tech Stack:** Python 3.13, Dagster 1.12.x, numpy, urllib (no new heavy deps), pytest, strict mypy.

**Spec:** `docs/superpowers/specs/2026-06-15-lifecycle-b1-paper-tracks-design.md`

**Reference (behavioural source, read-only):** foxhunt `scripts/surfer/{sixtyforty_paper,multistrat_paper,growth_discipline_paper,crypto_funding_paper,cross_venue_funding,surfer_poc}.py`

**Conventions for every task:** run tests with `cd ~/Work/fxhnt && .venv/bin/python -m pytest <path> -v`. Type-check with `.venv/bin/python -m mypy src/fxhnt`. Orchestration modules (`adapters/orchestration/*`) MUST NOT use `from __future__ import annotations` (Dagster runtime hint inspection); every other new module SHOULD. Commit after each green task with the trailer `Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>`.

---

## File Structure

| File | Responsibility |
|---|---|
| `src/fxhnt/application/forward_tracker.py` (modify) | Add `ForwardStrategy` Protocol + generic `ForwardTracker`; refactor `CombinedBookForwardTracker` onto it |
| `src/fxhnt/ports/market_data.py` (create) | Market-data client Protocols: `DailyBarClient`, `BinanceFundingClient`, `CrossVenueFundingClient`, `CryptoPerpBarClient`, `VolIndexClient` |
| `src/fxhnt/adapters/data/yahoo_daily.py` (create) | urllib `DailyBarClient` (Yahoo adjusted closes) |
| `src/fxhnt/adapters/data/binance_funding.py` (create) | urllib `BinanceFundingClient` |
| `src/fxhnt/adapters/data/cross_venue.py` (create) | urllib `CrossVenueFundingClient` (Binance/Bybit/Hyperliquid) |
| `src/fxhnt/adapters/data/crypto_perp.py` (create) | urllib `CryptoPerpBarClient` (Binance perp panel) |
| `src/fxhnt/adapters/data/vol_index.py` (create) | urllib `VolIndexClient` (Deribit DVOL) |
| `src/fxhnt/domain/strategies/sixtyforty.py` (create) | pure 60/40 daily return |
| `src/fxhnt/domain/strategies/multistrat.py` (create) | pure multi-asset adaptive book series |
| `src/fxhnt/domain/strategies/growth_discipline.py` (create) | pure GD blend (composes multistrat) |
| `src/fxhnt/domain/strategies/funding_carry.py` (create) | pure funding-qualify + book-carry |
| `src/fxhnt/domain/strategies/cross_venue.py` (create) | pure cross-venue spread book |
| `src/fxhnt/domain/strategies/momentum_vrp.py` (create) | pure residual-momentum weights + VRP sleeve |
| `src/fxhnt/application/paper_strategies.py` (create) | the 6 `ForwardStrategy` services wiring domain + data ports |
| `src/fxhnt/adapters/orchestration/assets.py` (modify) | 6 new `@asset`s + `cockpit_forward` deps |
| `src/fxhnt/adapters/orchestration/definitions.py` (modify) | register the 6 assets in job + Definitions |
| `tests/integration/test_forward_tracker.py` (create) | generic tracker behaviour |
| `tests/integration/test_paper_strategies.py` (create) | per-strategy services with fake data ports + round-trip through `state_reader` |
| foxhunt `infra/k8s/jobs/*surfer*` / cron manifests (modify) | retire migrated crons |

---

## Task 1: Generic ForwardTracker + ForwardStrategy port

**Files:**
- Modify: `src/fxhnt/application/forward_tracker.py`
- Test: `tests/integration/test_forward_tracker.py`

The existing `CombinedBookForwardTracker.step()` hard-codes `book.build()`. Generalize: a `ForwardStrategy` supplies candidate `(date, ret)` rows + an opaque `extra` carry-state; the tracker owns the days-list state file, the first-run inception freeze (book nothing), and forward-only booking.

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

```python
# tests/integration/test_forward_tracker.py
"""Generic ForwardTracker: first-run freeze, forward-only booking, idempotency, extra round-trip."""
from __future__ import annotations

from fxhnt.application.forward_tracker import ForwardTracker


class _Recomputable:
    """Recomputable-series strategy: always returns the full known series."""
    def __init__(self, series: list[tuple[str, float]]) -> None:
        self._series = series

    def advance(self, last_date, extra):
        return list(self._series), extra


class _LiveBooking:
    """Live-booking strategy: books one row for `today`, carries a counter in extra."""
    def __init__(self, today: str, ret: float) -> None:
        self._today, self._ret = today, ret

    def advance(self, last_date, extra):
        n = int(extra.get("runs", 0)) + 1
        return [(self._today, self._ret)], {"runs": n}


def test_first_run_freezes_and_books_nothing(tmp_path) -> None:
    p = str(tmp_path / "s.json")
    st = ForwardTracker(_Recomputable([("2026-01-01", 0.01), ("2026-01-02", 0.02)]), p).step()
    assert st.forward_days == 0                 # freeze: nothing booked on first run
    assert st.inception == "2026-01-02"         # inception = latest known date
    assert st.last_date == "2026-01-02"


def test_second_run_books_only_forward_days(tmp_path) -> None:
    p = str(tmp_path / "s.json")
    strat = _Recomputable([("2026-01-01", 0.01), ("2026-01-02", 0.02)])
    ForwardTracker(strat, p).step()             # freeze at 2026-01-02
    strat._series.append(("2026-01-03", 0.05))  # a new day appears
    st = ForwardTracker(strat, p).step()
    assert st.forward_days == 1                  # only 2026-01-03 booked
    assert abs(st.forward_return_pct - 5.0) < 1e-9
    assert st.last_date == "2026-01-03"


def test_idempotent_rerun_books_nothing(tmp_path) -> None:
    p = str(tmp_path / "s.json")
    strat = _Recomputable([("2026-01-01", 0.01), ("2026-01-02", 0.02), ("2026-01-03", 0.05)])
    ForwardTracker(strat, p).step()             # freeze at 2026-01-03
    st = ForwardTracker(strat, p).step()        # nothing new
    assert st.forward_days == 0


def test_extra_carry_state_round_trips(tmp_path) -> None:
    p = str(tmp_path / "s.json")
    ForwardTracker(_LiveBooking("2026-01-01", 0.0), p).step()         # freeze, extra={"runs":1}
    ForwardTracker(_LiveBooking("2026-01-02", 0.01), p).step()        # books 2026-01-02
    import json
    extra = json.load(open(p))["extra"]
    assert extra["runs"] == 2                    # advance saw prior extra and incremented
```

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

Run: `.venv/bin/python -m pytest tests/integration/test_forward_tracker.py -v`
Expected: FAIL (`ForwardTracker` does not exist).

- [ ] **Step 3: Implement**

Append to `src/fxhnt/application/forward_tracker.py` (keep the existing imports + `ForwardStatus`):

```python
import datetime as _dt
from typing import Protocol


class ForwardStrategy(Protocol):
    def advance(self, last_date: str | None, extra: dict) -> tuple[list[tuple[str, float]], dict]:
        """Return candidate (ISO-date, fractional-return) rows the strategy knows now, plus its updated
        carry-state. The tracker books only rows strictly after last_date and persists `extra` opaquely."""
        ...


def _today_iso() -> str:
    return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%d")


class ForwardTracker:
    """Strategy-agnostic forward-validation tracker. Freezes inception on first run (books nothing — no
    backfill), then books only days strictly after last_date. Owns the days-list state file."""

    def __init__(self, strategy: ForwardStrategy, state_path: str) -> None:
        self._strategy = strategy
        self._path = state_path

    def _load(self) -> dict | None:
        if os.path.exists(self._path):
            with open(self._path) as f:
                return json.load(f)
        return None

    def _save(self, state: dict) -> None:
        os.makedirs(os.path.dirname(self._path) or ".", exist_ok=True)
        with open(self._path, "w") as f:
            json.dump(state, f, indent=2)

    def step(self) -> ForwardStatus:
        state = self._load()
        first = state is None
        prev_last = None if first else (state["last_date"] or None)
        prev_extra = {} if first else state.get("extra", {})
        rows, new_extra = self._strategy.advance(prev_last, prev_extra)
        rows = sorted(rows, key=lambda x: x[0])

        booked = 0
        if first:
            latest = rows[-1][0] if rows else _today_iso()
            state = {"inception": latest, "last_date": latest, "nav": 1.0, "days": [], "extra": new_extra}
        else:
            for d, r in rows:
                if d > state["last_date"]:
                    state["nav"] *= (1.0 + r)
                    state["days"].append({"date": d, "ret": r, "nav": state["nav"]})
                    state["last_date"] = d
                    booked += 1
            state["extra"] = new_extra
        self._save(state)

        rets = np.array([row["ret"] for row in state["days"]])
        nav = np.array([row["nav"] for row in state["days"]])
        sharpe = float(per_period_sharpe(rets) * _ANN) if len(rets) >= 3 else 0.0
        mdd = float((nav / np.maximum.accumulate(nav) - 1.0).min()) if len(nav) else 0.0
        return ForwardStatus(
            inception=state["inception"], last_date=state["last_date"], forward_days=len(state["days"]),
            forward_return_pct=100.0 * (state["nav"] - 1.0), forward_sharpe=sharpe,
            forward_max_drawdown=mdd, booked_today=booked)
```

Then refactor `CombinedBookForwardTracker` to delegate (keeps call sites + existing tests working):

```python
class CombinedBookStrategy:
    """Recomputable-series ForwardStrategy over a CombinedBook."""
    def __init__(self, book: CombinedBook) -> None:
        self._book = book

    def advance(self, last_date: str | None, extra: dict) -> tuple[list[tuple[str, float]], dict]:
        _, dates, combo = self._book.build()
        rows = [(d, float(r)) for d, r in zip(dates, combo) if np.isfinite(r)]
        return rows, extra


def CombinedBookForwardTracker(book: CombinedBook, state_path: str) -> ForwardTracker:  # noqa: N802
    """Back-compat factory: the combined book is just a recomputable ForwardStrategy."""
    return ForwardTracker(CombinedBookStrategy(book), state_path)
```

Delete the old `CombinedBookForwardTracker` class body (its `_load`/`_save`/`step` now live on `ForwardTracker`).

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

Run: `.venv/bin/python -m pytest tests/integration/test_forward_tracker.py tests/integration/test_orchestration_assets.py -v`
Expected: PASS (new tracker tests + existing combined-asset tests stay green).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/forward_tracker.py tests/integration/test_forward_tracker.py
git commit -m "feat(b1): generic ForwardTracker over a ForwardStrategy port"
```

---

## Task 2: Market-data ports

**Files:**
- Create: `src/fxhnt/ports/market_data.py`

No test (Protocols only — exercised by every later task's fakes).

- [ ] **Step 1: Create the ports**

```python
# src/fxhnt/ports/market_data.py
"""Market-data client ports for the paper-forward strategies. Each has a urllib live adapter
(adapters/data/*) and a deterministic fake in tests. No network in the domain or in tests."""
from __future__ import annotations

from typing import Protocol


class DailyBarClient(Protocol):
    def adj_closes(self, symbol: str) -> dict[str, float]:
        """{ 'YYYY-MM-DD': adjusted_close } for a daily-bar history window."""
        ...


class BinanceFundingClient(Protocol):
    def universe(self) -> list[str]: ...
    def funding_daily(self, symbol: str) -> float:
        """Most-recent funding as a per-DAY rate (sum of the day's intervals)."""
        ...
    def quote_volume_usd(self, symbol: str) -> float: ...


class CrossVenueFundingClient(Protocol):
    def funding_by_venue(self) -> dict[str, dict[str, float]]:
        """{ coin: { venue: daily_funding_rate } } across the supported venues."""
        ...
    def volume_by_venue(self) -> dict[str, dict[str, float]]:
        """{ coin: { venue: 24h_quote_volume_usd } }."""
        ...


class CryptoPerpBarClient(Protocol):
    def panel(self) -> tuple[list[str], list[int], list[list[float]], list[list[float]], list[float]]:
        """(symbols, epoch_days, close[d][s], open[d][s], funding_now[s]) for the top-liquid perps."""
        ...


class VolIndexClient(Protocol):
    def dvol(self, currency: str) -> dict[int, float]:
        """{ epoch_day: implied-vol-index } for BTC/ETH (Deribit DVOL)."""
        ...
```

- [ ] **Step 2: Type-check**

Run: `.venv/bin/python -m mypy src/fxhnt/ports/market_data.py`
Expected: no errors.

- [ ] **Step 3: Commit**

```bash
git add src/fxhnt/ports/market_data.py
git commit -m "feat(b1): market-data client ports"
```

---

## Task 3: sixtyforty (TEMPLATE — domain + service + Yahoo adapter + fake)

This task establishes the pattern every recomputable Yahoo strategy follows. **Source:** foxhunt `scripts/surfer/sixtyforty_paper.py` (`series()`/`cmd_run()` — daily-rebalanced `0.6·r_SPY + 0.4·r_IEF` on adjusted closes).

**Files:**
- Create: `src/fxhnt/domain/strategies/sixtyforty.py`
- Create: `src/fxhnt/adapters/data/yahoo_daily.py`
- Modify: `src/fxhnt/application/paper_strategies.py` (create with the first service)
- Test: `tests/integration/test_paper_strategies.py` (create)

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

```python
# tests/integration/test_paper_strategies.py
"""Paper-forward strategy services: deterministic domain + service over fake data ports, and a
round-trip through ForwardStateReader so the cockpit contract is verified."""
from __future__ import annotations

from fxhnt.adapters.persistence.state_reader import ForwardStateReader
from fxhnt.application.forward_tracker import ForwardTracker
from fxhnt.application.paper_strategies import SixtyFortyStrategy


class FakeDailyBars:
    def __init__(self, data: dict[str, dict[str, float]]) -> None:
        self._data = data

    def adj_closes(self, symbol: str) -> dict[str, float]:
        return self._data[symbol]


def test_sixtyforty_daily_return_is_weighted_mean() -> None:
    from fxhnt.domain.strategies.sixtyforty import daily_returns
    closes = {
        "SPY": {"2026-01-01": 100.0, "2026-01-02": 110.0},   # +10%
        "IEF": {"2026-01-01": 100.0, "2026-01-02": 105.0},   # +5%
    }
    rows = daily_returns(closes, [("SPY", 0.6), ("IEF", 0.4)])
    assert rows == [("2026-01-02", 0.6 * 0.10 + 0.4 * 0.05)]  # 0.08


def test_sixtyforty_service_round_trips_through_reader(tmp_path) -> None:
    bars = FakeDailyBars({
        "SPY": {"2026-01-01": 100.0, "2026-01-02": 110.0, "2026-01-03": 110.0},
        "IEF": {"2026-01-01": 100.0, "2026-01-02": 105.0, "2026-01-03": 105.0},
    })
    p = str(tmp_path / "sixtyforty_state.json")
    ForwardTracker(SixtyFortyStrategy(bars), p).step()       # freeze at 2026-01-03
    # add a forward day
    bars._data["SPY"]["2026-01-04"] = 121.0                  # +10%
    bars._data["IEF"]["2026-01-04"] = 105.0                  # 0%
    st = ForwardTracker(SixtyFortyStrategy(bars), p).step()
    assert st.forward_days == 1
    summary, rows = ForwardStateReader().read(p, "sixtyforty")
    assert summary.days == 1
    assert abs(summary.nav - (1.0 + 0.06)) < 1e-9            # 0.6*0.10 + 0.4*0
```

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

Run: `.venv/bin/python -m pytest tests/integration/test_paper_strategies.py -v`
Expected: FAIL (modules missing).

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

```python
# src/fxhnt/domain/strategies/sixtyforty.py
"""Pure 60/40 daily-rebalanced return: for each consecutive date pair, the weighted mean of per-asset
returns on adjusted closes. No I/O."""
from __future__ import annotations


def daily_returns(closes: dict[str, dict[str, float]], weights: list[tuple[str, float]]) -> list[tuple[str, float]]:
    syms = [s for s, _ in weights]
    dates = sorted(set.intersection(*[set(closes[s]) for s in syms]))
    out: list[tuple[str, float]] = []
    for i in range(1, len(dates)):
        d, dprev = dates[i], dates[i - 1]
        r = sum(w * (closes[s][d] / closes[s][dprev] - 1.0) for s, w in weights)
        out.append((d, r))
    return out
```

- [ ] **Step 4: Implement the Yahoo adapter**

```python
# src/fxhnt/adapters/data/yahoo_daily.py
"""urllib DailyBarClient — Yahoo free daily adjusted closes. Retries with backoff; adjusted closes give
true total return (dividends + coupons). Mirrors foxhunt sixtyforty_paper.daily_adjclose."""
from __future__ import annotations

import datetime as dt
import json
import time
import urllib.request

_BASE = "https://query1.finance.yahoo.com/v8/finance/chart/{sym}?interval=1d&range={rng}"


class YahooDailyClient:
    def __init__(self, rng: str = "2y") -> None:
        self._rng = rng

    def _get(self, url: str, tries: int = 4) -> dict:
        for a in range(tries):
            try:
                req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
                return json.loads(urllib.request.urlopen(req, timeout=30).read())
            except Exception:  # noqa: BLE001 — transient HTTP; retry then re-raise
                if a == tries - 1:
                    raise
                time.sleep(2 * (a + 1))
        raise RuntimeError("unreachable")

    def adj_closes(self, symbol: str) -> dict[str, float]:
        res = self._get(_BASE.format(sym=symbol, rng=self._rng))["chart"]["result"][0]
        ts = res["timestamp"]
        ind = res["indicators"]
        adj = ind.get("adjclose", [{}])[0].get("adjclose") or ind["quote"][0]["close"]
        out: dict[str, float] = {}
        for t, c in zip(ts, adj):
            if c is not None:
                out[dt.datetime.utcfromtimestamp(t).strftime("%Y-%m-%d")] = float(c)
        return out
```

- [ ] **Step 5: Implement the service**

```python
# src/fxhnt/application/paper_strategies.py
"""The paper-forward ForwardStrategy services: each composes a market-data port + a pure domain module
and yields candidate (date, ret) rows for the generic ForwardTracker. Recomputable strategies return the
full series each run (the tracker books only the new tail); live-booking strategies book today + carry
positions in `extra`."""
from __future__ import annotations

from fxhnt.domain.strategies import sixtyforty
from fxhnt.ports.market_data import DailyBarClient

_SIXTYFORTY_WEIGHTS = [("SPY", 0.6), ("IEF", 0.4)]


class SixtyFortyStrategy:
    def __init__(self, bars: DailyBarClient) -> None:
        self._bars = bars

    def advance(self, last_date: str | None, extra: dict) -> tuple[list[tuple[str, float]], dict]:
        closes = {s: self._bars.adj_closes(s) for s, _ in _SIXTYFORTY_WEIGHTS}
        return sixtyforty.daily_returns(closes, _SIXTYFORTY_WEIGHTS), extra
```

- [ ] **Step 6: Run to verify it passes**

Run: `.venv/bin/python -m pytest tests/integration/test_paper_strategies.py -v && .venv/bin/python -m mypy src/fxhnt`
Expected: PASS, no mypy errors.

- [ ] **Step 7: Commit**

```bash
git add src/fxhnt/domain/strategies/sixtyforty.py src/fxhnt/adapters/data/yahoo_daily.py src/fxhnt/application/paper_strategies.py tests/integration/test_paper_strategies.py
git commit -m "feat(b1): sixtyforty paper-forward strategy (Yahoo) — template"
```

---

## Task 4: multistrat (adaptive multi-asset book)

**Source:** foxhunt `scripts/surfer/multistrat_paper.py` — port `build()`, `book_series()`, `yhist()` semantics EXACTLY. Six instruments `SPY/IEF/GLD/PDBC/DBMF/BTC-USD`; vol-normalize each stream; edge-decay trust θ (EMA on per-stream Sharpe); combine; continuous adaptive leverage = `min(MAXLEV, max(LEV_FLOOR, vol_target/realized_vol_ema)) · kelly_factor · dd_delever`; `TARGET_VOL=0.10`, `MAXLEV=1.0`, `KELLY_FLOOR=0.5`, `LEV_FLOOR=0.3`, `DD_DEADBAND=0.05`, `DD_SENS=3.0`, `DD_FLOOR=0.40`. Recomputable-series pattern.

**Files:**
- Create: `src/fxhnt/domain/strategies/multistrat.py` (pure: `book_series(closes: dict[str, dict[str, float]]) -> list[tuple[str, float]]`)
- Modify: `src/fxhnt/application/paper_strategies.py` (add `MultiStratStrategy`)
- Modify: `tests/integration/test_paper_strategies.py`

- [ ] **Step 1: Write the failing test** — deterministic synthetic closes (two trending + one flat instrument over ~40 days) and assert (a) the booked series length equals `len(dates)-warmup`, (b) every `nav` is finite and `>0`, (c) realized annualized vol of the booked returns is within ±25% of `TARGET_VOL=0.10` (the vol-targeting invariant), (d) leverage never exceeds `MAXLEV=1.0` (assert `max|daily ret| <= MAXLEV * max|underlying daily ret| + 1e-9`). Use a `FakeDailyBars` covering the 6 symbols.

```python
def test_multistrat_vol_targets_and_caps_leverage() -> None:
    from fxhnt.domain.strategies.multistrat import book_series, INSTRUMENTS
    import math, numpy as np
    rng = np.random.default_rng(0)
    dates = [f"2026-{m:02d}-{d:02d}" for m in (1, 2) for d in range(1, 29)]
    closes = {}
    for k, s in enumerate(INSTRUMENTS):
        steps = rng.normal(0.0003 * (1 if k % 2 else -1), 0.01, len(dates))
        px = 100.0 * np.cumprod(1 + steps)
        closes[s] = {d: float(px[i]) for i, d in enumerate(dates)}
    rows = book_series(closes)
    rets = np.array([r for _, r in rows])
    assert len(rows) >= 1 and np.isfinite(rets).all()
    vol = float(rets.std() * math.sqrt(252))
    assert 0.05 <= vol <= 0.15            # vol-targeted near 0.10 (band for finite sample)
```

- [ ] **Step 2: Run to verify it fails** — `.venv/bin/python -m pytest tests/integration/test_paper_strategies.py::test_multistrat_vol_targets_and_caps_leverage -v` → FAIL.

- [ ] **Step 3: Implement `src/fxhnt/domain/strategies/multistrat.py`** — port the foxhunt `build()`/`book_series()` math line-for-line into pure functions over `closes: dict[str, dict[str, float]]` (numpy internally). Expose module constant `INSTRUMENTS = ["SPY","IEF","GLD","PDBC","DBMF","BTC-USD"]` and `book_series(closes) -> list[tuple[str, float]]` returning `(date, daily_return)` for each booked day (skip the warmup window exactly as the original `yhist`/`build` do). Keep the adaptive-leverage block (vol EMA, Kelly floor, DD de-lever) identical to the source; no regime overlay (overlay OFF, matching the default).

- [ ] **Step 4: Add the service to `paper_strategies.py`**

```python
from fxhnt.domain.strategies import multistrat

class MultiStratStrategy:
    def __init__(self, bars: DailyBarClient) -> None:
        self._bars = bars

    def advance(self, last_date: str | None, extra: dict) -> tuple[list[tuple[str, float]], dict]:
        closes = {s: self._bars.adj_closes(s) for s in multistrat.INSTRUMENTS}
        return multistrat.book_series(closes), extra
```

- [ ] **Step 5: Run to verify it passes** — pytest (the new test) + mypy. Expected: PASS.

- [ ] **Step 6: Commit** — `git commit -m "feat(b1): multistrat adaptive paper-forward book (Yahoo)"`

---

## Task 5: gd (growth-discipline blend, composes multistrat)

**Source:** foxhunt `scripts/surfer/growth_discipline_paper.py` — `w·r_QQQ + (1−w)·r_multistrat`, `w = GD_EQUITY_WEIGHT` (default 0.70). Composes Task 4's `multistrat.book_series`. Recomputable.

**Files:**
- Create: `src/fxhnt/domain/strategies/growth_discipline.py`
- Modify: `src/fxhnt/application/paper_strategies.py` (add `GrowthDisciplineStrategy`)
- Modify: `tests/integration/test_paper_strategies.py`

- [ ] **Step 1: Write the failing test** — fake bars covering QQQ + the 6 multistrat instruments; assert `gd` daily return on a date equals `w·r_QQQ(date) + (1−w)·r_multistrat(date)` for the overlapping dates (reconstruct the expected from `multistrat.book_series` + the QQQ return directly), with `w=0.70`.

- [ ] **Step 2: Run to verify it fails** → FAIL.

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

```python
# src/fxhnt/domain/strategies/growth_discipline.py
"""Growth-discipline blend: w·QQQ + (1−w)·multistrat book, on the intersection of their dates. Pure."""
from __future__ import annotations

from fxhnt.domain.strategies import multistrat


def daily_returns(qqq: dict[str, float], multistrat_closes: dict[str, dict[str, float]],
                  w_equity: float = 0.70) -> list[tuple[str, float]]:
    ms = dict(multistrat.book_series(multistrat_closes))
    qdates = sorted(qqq)
    qret = {qdates[i]: qqq[qdates[i]] / qqq[qdates[i - 1]] - 1.0 for i in range(1, len(qdates))}
    common = sorted(set(ms) & set(qret))
    return [(d, w_equity * qret[d] + (1.0 - w_equity) * ms[d]) for d in common]
```

- [ ] **Step 4: Add the service** (reads `GD_EQUITY_WEIGHT` from env, default 0.70):

```python
import os
from fxhnt.domain.strategies import growth_discipline

class GrowthDisciplineStrategy:
    def __init__(self, bars: DailyBarClient) -> None:
        self._bars = bars

    def advance(self, last_date: str | None, extra: dict) -> tuple[list[tuple[str, float]], dict]:
        w = float(os.environ.get("GD_EQUITY_WEIGHT", "0.70"))
        qqq = self._bars.adj_closes("QQQ")
        closes = {s: self._bars.adj_closes(s) for s in multistrat.INSTRUMENTS}
        return growth_discipline.daily_returns(qqq, closes, w), extra
```

- [ ] **Step 5: Run to verify it passes** — pytest + mypy → PASS.
- [ ] **Step 6: Commit** — `git commit -m "feat(b1): growth-discipline paper-forward blend (composes multistrat)"`

---

## Task 6: BinanceFundingClient adapter + funding strategy

**Source:** foxhunt `scripts/surfer/crypto_funding_paper.py` — qualify coins with trailing funding `> HURDLE=5bp/day` and liquidity `> $5M`; equal-weight; each day book yesterday's positions' realized funding minus `COST_RT=10bp` on turnover; `EXIT_HURDLE=3bp`. **Live-booking** pattern: positions persist in `extra`. State emitted = days-list (`ret = today's realized carry`).

**Files:**
- Create: `src/fxhnt/adapters/data/binance_funding.py` (`BinanceFundingClient`)
- Create: `src/fxhnt/domain/strategies/funding_carry.py` (pure qualify + book)
- Modify: `src/fxhnt/application/paper_strategies.py` (add `FundingCarryStrategy`)
- Modify: `tests/integration/test_paper_strategies.py`

- [ ] **Step 1: Write the failing test** — a `FakeBinanceFunding` returning a fixed universe + funding/volume. Drive two `advance()` calls: first run sets positions, books nothing (returns `[]` or a today-row the tracker freezes); second run (next day) books `realized = mean(funding of held qualifiers) − cost·turnover`. Assert the booked return equals the hand-computed value and `extra["positions"]` holds the equal-weight book. Then round-trip through `ForwardStateReader` (days-list path) → `summary.nav == 1 + booked`.

```python
class FakeBinanceFunding:
    def __init__(self, fund: dict[str, float], vol: dict[str, float]) -> None:
        self._f, self._v = fund, vol
    def universe(self): return list(self._f)
    def funding_daily(self, s): return self._f[s]
    def quote_volume_usd(self, s): return self._v[s]
```

- [ ] **Step 2: Run to verify it fails** → FAIL.

- [ ] **Step 3: Implement the domain** `funding_carry.py` — pure functions: `qualify(funding, volume, hurdle=5e-4, exit_hurdle=3e-4, liq=5e6, held) -> list[str]` (hysteresis: keep held coins above exit_hurdle, add new above hurdle, all above liq); `book(prev_positions, funding, cost_rt=1e-3) -> float` (equal-weight realized carry on prev positions minus cost·turnover when rebalancing). Mirror the source constants exactly.

- [ ] **Step 4: Implement the adapter** `binance_funding.py` — urllib calls to `fapi.binance.com` (`/fapi/v1/exchangeInfo`, `/fapi/v1/fundingRate?limit=...`, `/fapi/v1/ticker/24hr`) reusing the same `_get` retry helper shape as `yahoo_daily.py`; `funding_daily` = sum of the most recent day's funding intervals (`INTERVALS_PER_DAY=3`).

- [ ] **Step 5: Implement the service** `FundingCarryStrategy.advance(last_date, extra)` — read `extra.get("positions", {})`; fetch funding/volume; compute today's realized `book(...)`; requalify + rebalance positions; return `([(today_iso, realized)], {"positions": new_positions})`. Use `_today_iso()` from `forward_tracker`.

- [ ] **Step 6: Run to verify it passes** — pytest + mypy → PASS.
- [ ] **Step 7: Commit** — `git commit -m "feat(b1): crypto funding-harvest paper-forward strategy (Binance)"`

---

## Task 7: CrossVenueFundingClient adapter + crossvenue strategy

**Source:** foxhunt `scripts/surfer/cross_venue_funding.py` — for each coin, short the highest-funding venue perp + long the lowest; book the spread; `ENTRY=10bp`, `EXIT=5bp` hysteresis, `TOPK=10`, `LIQ=$10M` both legs, `MAXF=50bp` distress filter, `COST_RT=10bp`. Live-booking. Venues: Binance/Bybit/Hyperliquid.

**Files:**
- Create: `src/fxhnt/adapters/data/cross_venue.py` (`CrossVenueFundingClient`)
- Create: `src/fxhnt/domain/strategies/cross_venue.py` (pure spread book)
- Modify: `src/fxhnt/application/paper_strategies.py` (add `CrossVenueStrategy`)
- Modify: `tests/integration/test_paper_strategies.py`

- [ ] **Step 1: Write the failing test** — `FakeCrossVenue` returning `funding_by_venue` + `volume_by_venue` for 3 coins; assert the spread book picks the top-K spreads with hysteresis and the booked daily return equals `mean(spread of held) − cost·turnover`; `extra["positions"]` round-trips. Round-trip through reader (days-list).

- [ ] **Step 2: Run to verify it fails** → FAIL.

- [ ] **Step 3: Implement domain** `cross_venue.py` — pure: `spreads(funding_by_venue, volume_by_venue, liq, maxf) -> dict[str,float]` (per-coin max−min funding across venues passing liquidity + distress filters); `select(spreads, held, entry, exit_, topk) -> list[str]`; `book(prev_positions, spreads, cost_rt) -> float`. Constants from source.

- [ ] **Step 4: Implement adapter** `cross_venue.py` (adapters/data) — urllib to `fapi.binance.com` (`/fapi/v1/premiumIndex`,`/fapi/v1/fundingInfo`,`/fapi/v1/ticker/24hr`), `api.bybit.com` (`/v5/market/tickers?category=linear`), `api.hyperliquid.xyz` (`POST /info {"type":"metaAndAssetCtxs"}`); normalize each venue's funding to a daily rate; assemble `{coin:{venue:rate}}` + volumes. Reuse the `_get` retry shape; POST helper for Hyperliquid.

- [ ] **Step 5: Implement the service** `CrossVenueStrategy.advance` — mirror Task 6's structure (positions in `extra`, return `[(today_iso, realized)]`).

- [ ] **Step 6: Run to verify it passes** — pytest + mypy → PASS.
- [ ] **Step 7: Commit** — `git commit -m "feat(b1): cross-venue funding paper-forward strategy"`

---

## Task 8: poc strategy — FAITHFUL port WITH torch (vendor the real helpers)

**CORRECTION (do not dodge deps):** poc is ported **faithfully** — keep the algorithm 1:1. Use **torch**; use the **real** `signal_sweep` (`xs_weights`, `validate`, `sharpe_t`) and **real** `pit_sweep` (`load()`); keep the **real** `_vrp_calib` (PIT-history-based VRP sizing). Do NOT reimplement any of this in numpy and do NOT change the calibration source — changing an algorithm to avoid a package is not allowed. The only wrapping is persistence: the live mark/rebalance computation is unchanged; the tracker books `comb_ret` into days-list and the bespoke poc fields live in `extra`.

**Source:** foxhunt `scripts/surfer/surfer_poc.py` (`paper(cfg)`, `compute_weights`, `vrp_ccy`/`vrp_book_series`, `_vrp_calib`, `_live_panel`), `scripts/surfer/signal_sweep.py`, `scripts/surfer/pit_sweep.py`.

**Files:**
- Create: `src/fxhnt/vendor/surfer/__init__.py`, `signal_sweep.py`, `pit_sweep.py`, `surfer_poc.py` — copied VERBATIM from foxhunt `scripts/surfer/` (preserve the algorithm). Only allowed edits: make the state path + the PIT-cache path configurable via env/param instead of the hardcoded `_REPO`-relative paths (so they resolve to `/data/surfer` + the pod's PIT cache); keep all math, torch usage, and the `from signal_sweep/pit_sweep import ...` wiring intact (adjust to package-relative imports).
- Create: `src/fxhnt/adapters/data/crypto_perp.py` (`CryptoPerpBarClient`) + `src/fxhnt/adapters/data/vol_index.py` (`VolIndexClient`) — only if the vendored `_live_panel`/`_dvol` are refactored to take injected clients; otherwise the vendored module keeps its own urllib fetch (faithful) and these ports are skipped. Prefer the faithful path: keep the vendored fetch.
- Modify: `src/fxhnt/application/paper_strategies.py` (add `MomentumVrpStrategy` wrapping the vendored poc paper step).
- Modify: `pyproject.toml` (add `torch` to the orchestration/poc extra) and `infra/docker/fxhnt.Dockerfile` (install the extra so the cockpit image has torch). torch CPU wheel is sufficient (paper inference is tiny).
- Modify: `tests/integration/test_paper_strategies.py`.

- [ ] **Step 1: Vendor the real modules** — copy `signal_sweep.py`, `pit_sweep.py`, `surfer_poc.py` from foxhunt `scripts/surfer/` into `src/fxhnt/vendor/surfer/`. Convert the `sys.path.insert` + bare `from signal_sweep import ...` / `import pit_sweep` to package-relative imports (`from fxhnt.vendor.surfer.signal_sweep import ...`). Make `STATE`, the dvol cache path, and the pit-cache path read from env (`FXHNT_SURFER_DATA_DIR`, default `/data/surfer`; PIT cache path via the existing pit_sweep config) — NO other edits. Keep `import torch`.

- [ ] **Step 2: Write the failing test** (uses the REAL functions; torch on CPU; small fixtures):
  - `test_xs_weights_market_neutral_unit_gross`: real `signal_sweep.xs_weights` on a small `sig` → rows sum ≈0, `sum|w|` ≈1.
  - `test_compute_weights_shape_and_neutral`: real `surfer_poc.compute_weights` on a synthetic 50-perp × 80-day panel → ≤`topk` nonzeros/row, row-sum ≈0.
  - `test_momentum_vrp_service_round_trips`: drive `MomentumVrpStrategy` over a fixed local panel/DVOL fixture (monkeypatch the vendored fetch to return fixtures so no network) through two `ForwardTracker` steps → second step books one row; `ForwardStateReader().read(...)` returns a days-list summary with `days==1`; `extra` carries `positions`/`vrp_*`.

- [ ] **Step 3: Run to verify they fail** → FAIL.

- [ ] **Step 4: Implement the service** `MomentumVrpStrategy.advance(last_date, extra)` — call the vendored poc mark/rebalance logic (the real `paper`-equivalent, refactored to return values rather than print): seed its working state from `extra`; compute `comb_ret` exactly as `surfer_poc.paper` does (momentum mark + real `_vrp_calib`-sized VRP sleeve); return `([(iso(today), comb_ret)], {positions, prices, vrp_equity, vrp_mu, vrp_vu})`. Convert epoch-day → ISO date for the days-list. The numbers must match the vendored algorithm exactly.

- [ ] **Step 5: Image deps** — add `torch` to the relevant extra in `pyproject.toml`; update `infra/docker/fxhnt.Dockerfile` to install it; ensure the PIT cache `pit_sweep.load()` needs is present in the cockpit image or mounted (point `pit_sweep` at the warehouse/PVC path available to the Dagster pod). Note this as the one infra add for B1.

- [ ] **Step 6: Run to verify it passes** — pytest + mypy → PASS.
- [ ] **Step 7: Commit** — `git commit -m "feat(b1): surfer poc paper-forward strategy — faithful port with torch + real signal_sweep/pit_sweep"`

---

## Task 9: Orchestration — 6 assets + wiring

**Files:**
- Modify: `src/fxhnt/adapters/orchestration/assets.py`
- Modify: `src/fxhnt/adapters/orchestration/definitions.py`
- Test: `tests/integration/test_orchestration_definitions.py` (extend)

- [ ] **Step 1: Write the failing test** — extend `test_orchestration_definitions.py` to assert the asset keys now include `sixtyforty_nav, multistrat_nav, gd_nav, funding_nav, crossvenue_nav, poc_nav` alongside the existing four, and that `cockpit_forward`'s upstream deps include all six.

```python
assert {"sixtyforty_nav","multistrat_nav","gd_nav","funding_nav","crossvenue_nav","poc_nav"} <= asset_keys
```

- [ ] **Step 2: Run to verify it fails** → FAIL.

- [ ] **Step 3: Implement the assets** — in `assets.py` add one `@asset` per strategy following the `combined_forward_nav` shape. Each builds its service from live adapters + settings, runs `ForwardTracker(service, f"{_data_dir()}/{state_file}.json").step()`, logs `forward_days`/`last_date`, returns a small status dict, and is best-effort (wrap the live fetch so a transient API error logs a warning and the asset still steps/keeps state). State-file names MUST match the registry: `sixtyforty_state`, `multistrat_state`, `gd_paper_state`, `funding_paper_state`, `crossvenue_state`, `poc_state`. Example:

```python
@asset
def sixtyforty_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    from fxhnt.adapters.data.yahoo_daily import YahooDailyClient
    from fxhnt.application.forward_tracker import ForwardTracker
    from fxhnt.application.paper_strategies import SixtyFortyStrategy
    st = ForwardTracker(SixtyFortyStrategy(YahooDailyClient()),
                        f"{_data_dir()}/sixtyforty_state.json").step()
    context.log.info(f"sixtyforty_nav: {st.forward_days}d through {st.last_date}")
    return {"forward_days": st.forward_days, "last_date": st.last_date}
```

- [ ] **Step 4: Wire deps + Definitions** — set `cockpit_forward`'s decorator to `@asset(deps=[combined_forward_nav, sixtyforty_nav, multistrat_nav, gd_nav, funding_nav, crossvenue_nav, poc_nav])`. In `definitions.py` add the six assets to both `Definitions(assets=[...])` and the `define_asset_job(selection=[...])`.

- [ ] **Step 5: Run to verify it passes** — `.venv/bin/python -m pytest tests/integration/test_orchestration_definitions.py tests/integration/test_orchestration_assets.py -v` → PASS.
- [ ] **Step 6: Commit** — `git commit -m "feat(b1): wire 6 paper-track assets into the Dagster graph"`

---

## Task 10: Full suite, deploy, cron retirement, verify

**Files:**
- Modify: foxhunt cron/job manifests for the migrated surfer harnesses (suspend).

- [ ] **Step 1: Full local gate** — `cd ~/Work/fxhnt && SQLX_OFFLINE=true .venv/bin/python -m pytest -q && .venv/bin/python -m mypy src/fxhnt`. Expected: all green, no mypy errors. Commit any fixes.

- [ ] **Step 2: Push fxhnt** — `GIT_SSH_COMMAND="ssh -o BatchMode=yes -o StrictHostKeyChecking=accept-new -p 2222" git push origin master`.

- [ ] **Step 3: Build + deploy the cockpit image** — submit the `fxhnt-cockpit` Argo build at the new master HEAD (the now-green pipeline: `kubectl create` a Workflow with `workflowTemplateRef: {name: fxhnt-cockpit}` + `commit-sha=<HEAD>`); wait for `Succeeded`; confirm both nodes archive logs (the B0/CI fix).

- [ ] **Step 4: Trigger one materialization** — run the `combined_book_forward_job` once via the Dagster daemon/schedule (or `kubectl exec` the daemon run-launcher); confirm the 6 new state files appear in `/data/surfer` and `cockpit_forward` ingests them.

- [ ] **Step 5: Verify the cockpit** — query the cockpit Postgres: `SELECT strategy_id, days, gate_status FROM forward_summary ORDER BY strategy_id;` Expected: all 7 strategies present (`combined` + the 6), each with a fresh `src_updated_at`. The 6 new ones show `days=0`/`WAIT` on first run (inception freeze) — that is correct.

- [ ] **Step 6: Retire the foxhunt surfer crons** — in the foxhunt repo, suspend the cron/job manifests that ran the migrated harnesses (notably the funding cron at 01:17 UTC): set `spec.suspend: true`, commit, `kubectl apply`. Reversible. Note in the commit that the 2026-07-05 funding Phase-1 review still applies (same data, now Dagster-produced).

- [ ] **Step 7: Update memory** — update `project_lifecycle_b0_state.md` (or add a B1 note) recording: 6 strategies ported, all 7 live in cockpit, foxhunt surfer crons retired, poc done in numpy (no torch).

---

## Self-Review

**Spec coverage:** foundation tracker (Task 1) ✓; 5 data ports (Tasks 2,3,6,7,8) ✓; days-list unification (every service emits via the tracker → `_days_list` reader, verified by round-trip tests) ✓; all 6 strategies (Tasks 3,4,5,6,7,8) ✓; parallel assets / no DuckDB contention (Task 9 — assets touch no warehouse) ✓; no netpol change (noted; nothing to do) ✓; cron retirement + fresh inception (Task 10) ✓; poc numpy-only, paper-path-only, calibration-from-lookback (Task 8) ✓; testing with fakes, no live network, strict mypy (every task) ✓.

**Placeholder scan:** the heavy domain ports (multistrat Task 4, funding Task 6, crossvenue Task 7, poc Task 8) cite the exact foxhunt source function + the constants + the target signature + concrete tests with hand-computed expected values — the implementer ports cited math into a given signature verified by given assertions (not a vague "implement the logic"). Foundation, ports, sixtyforty, gd, and orchestration carry full literal code.

**Type consistency:** `ForwardStrategy.advance(last_date: str | None, extra: dict) -> tuple[list[tuple[str, float]], dict]` is used identically in Task 1 and every service (Tasks 3–8). `DailyBarClient.adj_closes` / `BinanceFundingClient.{universe,funding_daily,quote_volume_usd}` / `CrossVenueFundingClient.{funding_by_venue,volume_by_venue}` / `CryptoPerpBarClient.panel` / `VolIndexClient.dvol` match between the Task-2 ports, the adapters, and the services. State-file basenames in Task 9 match `STRATEGY_REGISTRY` exactly.
