# Ongoing Diversification-Aware Edge-Factory 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.

**Goal:** Build the continuous factory that finds uncorrelated edges and stacks them: analyst-fleet → gauntlet (multiple-testing-honest) → diversification gate (left-tail correlation + marginal contribution) → paper-forward → autonomous promotion → HRP allocation → edge-decay retirement, in-cluster daily, visible in the cockpit.

**Architecture:** Hexagonal (fxhnt), reusing the existing factory skeleton (`gauntlet`, `DiscoveryService`, `ForwardValidationService`, `StrategyStore`). New pure `domain/diversification/` core (correlation/HRP/marginal/gate); a `factory_loop` orchestrator that inserts the diversification gate into promotion + does HRP re-allocation + governance; a domain-analyst fleet feeding the same gauntlet; a cockpit funnel view; a daily Argo CronJob.

**Tech Stack:** Python 3.11, numpy, scipy (HRP clustering), SQLAlchemy, typer, pytest; ccxt/argo/k8s for deploy. Repo `/home/jgrusewski/Work/fxhnt`, branch `feat/edge-factory` (create from master). Commit 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 unless asked. Always `cd /home/jgrusewski/Work/fxhnt` (cwd may reset).

---

## File Structure

**Create:**
- `src/fxhnt/domain/diversification/__init__.py`
- `src/fxhnt/domain/diversification/correlation.py` — full-sample + left-tail correlation.
- `src/fxhnt/domain/diversification/hrp.py` — Hierarchical Risk Parity weights.
- `src/fxhnt/domain/diversification/marginal.py` — marginal Sharpe contribution.
- `src/fxhnt/domain/diversification/gate.py` — diversification admit/reject gate.
- `src/fxhnt/application/factory/__init__.py`
- `src/fxhnt/application/factory/governance.py` — kill-switch + caps.
- `src/fxhnt/application/factory/factory_loop.py` — the daily cycle orchestrator.
- `src/fxhnt/application/factory/analyst_fleet.py` — domain-analyst orchestration.
- `src/fxhnt/application/factory/analysts.py` — specialist analyst agents + a fake for tests.
- `src/fxhnt/adapters/persistence/factory_store.py` — funnel + allocation persistence (cockpit DB).
- tests under `tests/unit/` and `tests/integration/`.
- `infra/k8s/jobs/fxhnt-factory-cronjob.yaml`

**Modify:**
- `pyproject.toml` — add `factory` optional-dependency group (scipy).
- `src/fxhnt/config.py` — add `FactorySettings`.
- `src/fxhnt/adapters/persistence/cockpit_models.py` — add funnel/allocation rows.
- `src/fxhnt/cli.py` — `factory-cycle` / `factory-flatten` / `factory-status`.
- `src/fxhnt/adapters/web/` — cockpit funnel view.

---

# Milestone 1 — Diversification core (pure)

### Task 1: scipy dep + FactorySettings

**Files:** Modify `pyproject.toml`, `src/fxhnt/config.py`; Test `tests/unit/test_factory_settings.py`

- [ ] **Step 1: Add the factory dep group**

In `pyproject.toml` under `[project.optional-dependencies]` add:
```toml
factory = ["scipy>=1.11"]
```

- [ ] **Step 2: Install + verify** — `cd /home/jgrusewski/Work/fxhnt && pip install -e '.[dev,factory]' && python -c "import scipy.cluster.hierarchy; print('scipy ok')"` → prints `scipy ok`.

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

Create `tests/unit/test_factory_settings.py`:
```python
"""FactorySettings: safe diversification + governance defaults."""
from __future__ import annotations

from fxhnt.config import FactorySettings


def test_safe_defaults() -> None:
    s = FactorySettings()
    assert 0.0 < s.max_tail_corr < 1.0
    assert s.min_marginal_sharpe >= 0.0
    assert s.tail_frac > 0.0 and s.tail_frac <= 0.5
    assert s.max_deployed_sleeves >= 1
    assert s.max_new_deploys_per_cycle >= 1
    assert s.kill_switch is False
    assert s.min_forward_days >= 10
```

- [ ] **Step 4: Run (fails)** — `pytest tests/unit/test_factory_settings.py -v` → ImportError.

- [ ] **Step 5: Add FactorySettings**

In `src/fxhnt/config.py` after `BinanceSettings`, add:
```python
class FactorySettings(BaseSettings):
    """Ongoing edge-factory: diversification gate + governance. Conservative by design."""
    model_config = SettingsConfigDict(env_prefix="FXHNT_FACTORY_")
    max_tail_corr: float = 0.5         # admit a sleeve only if its LEFT-TAIL corr to the book is below this
    min_marginal_sharpe: float = 0.05  # ...and it adds at least this much to book Sharpe
    tail_frac: float = 0.10            # "left tail" = the book's worst 10% of days
    min_forward_days: int = 20         # forward observations before a sleeve can promote
    max_deployed_sleeves: int = 20     # cap on the deployed book size
    max_new_deploys_per_cycle: int = 2 # a burst can't over-commit capital in one cycle
    kill_switch: bool = False          # halt all new promotions
```
Register on `Settings` beside `binance`: `factory: "FactorySettings" = Field(default_factory=FactorySettings)`.

- [ ] **Step 6: Run (passes) + commit**
```bash
git add pyproject.toml src/fxhnt/config.py tests/unit/test_factory_settings.py
git commit -m "feat(factory): scipy dep + FactorySettings (safe diversification/governance defaults)"
```

---

### Task 2: Correlation — full-sample + left-tail

**Files:** Create `src/fxhnt/domain/diversification/__init__.py` (empty), `correlation.py`; Test `tests/unit/test_correlation.py`

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

Create `tests/unit/test_correlation.py`:
```python
"""Full-sample and LEFT-TAIL correlation. Left-tail = correlation on the book's worst days (crisis-relevant)."""
from __future__ import annotations

import numpy as np

from fxhnt.domain.diversification.correlation import full_correlation, left_tail_correlation


def test_full_correlation_aligned() -> None:
    a = np.array([1.0, -1.0, 2.0, -2.0, 0.5])
    assert abs(full_correlation(a, a) - 1.0) < 1e-9
    assert abs(full_correlation(a, -a) + 1.0) < 1e-9


def test_left_tail_picks_book_worst_days() -> None:
    # book: last two days are the worst; sleeve co-moves ONLY on those days -> high left-tail corr,
    # near-zero full-sample corr.
    rng = np.random.default_rng(0)
    n = 200
    book = rng.normal(0, 0.01, n)
    book[-2:] = [-0.05, -0.06]                    # crisis days
    sleeve = rng.normal(0, 0.01, n)
    sleeve[-2:] = [-0.05, -0.07]                  # sleeve crashes WITH the book
    lt = left_tail_correlation(sleeve, book, tail_frac=0.02)   # worst ~4 days
    fs = full_correlation(sleeve, book)
    assert lt > 0.5                                # co-moves in the tail
    assert lt > fs                                 # tail corr exceeds full-sample (the crisis-hiding case)


def test_left_tail_unestimable_returns_nan() -> None:
    book = np.array([0.01, 0.02])                 # too few days for a tail
    sleeve = np.array([0.0, 0.0])
    assert np.isnan(left_tail_correlation(sleeve, book, tail_frac=0.1))
```

- [ ] **Step 2: Run (fails)** — `pytest tests/unit/test_correlation.py -v` → ModuleNotFound.

- [ ] **Step 3: Implement**

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

Create `src/fxhnt/domain/diversification/correlation.py`:
```python
"""Correlation between return streams — full-sample and LEFT-TAIL. The left tail (the book's worst days) is
the crisis-relevant measure: correlations spike in selloffs, so a sleeve uncorrelated full-sample but
co-moving in the tail must be caught. Returns NaN when there are too few aligned/tail days to estimate."""
from __future__ import annotations

import numpy as np

_MIN_TAIL_DAYS = 3


def _aligned(a: np.ndarray, b: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
    n = min(len(a), len(b))
    a, b = a[-n:], b[-n:]
    m = np.isfinite(a) & np.isfinite(b)
    return a[m], b[m]


def full_correlation(a: np.ndarray, b: np.ndarray) -> float:
    a, b = _aligned(a, b)
    if len(a) < 2 or a.std() == 0 or b.std() == 0:
        return float("nan")
    return float(np.corrcoef(a, b)[0, 1])


def left_tail_correlation(sleeve: np.ndarray, book: np.ndarray, *, tail_frac: float = 0.10) -> float:
    """Correlation of sleeve vs book on the days the BOOK is in its worst `tail_frac` quantile."""
    s, b = _aligned(sleeve, book)
    k = int(len(b) * tail_frac)
    if k < _MIN_TAIL_DAYS:
        return float("nan")
    worst = np.argsort(b)[:k]                      # indices of the book's lowest returns
    sv, bv = s[worst], b[worst]
    if sv.std() == 0 or bv.std() == 0:
        return float("nan")
    return float(np.corrcoef(sv, bv)[0, 1])
```

- [ ] **Step 4: Run (passes) + commit**
```bash
git add src/fxhnt/domain/diversification/__init__.py src/fxhnt/domain/diversification/correlation.py tests/unit/test_correlation.py
git commit -m "feat(factory): full-sample + left-tail correlation (crisis-relevant)"
```

---

### Task 3: Hierarchical Risk Parity weights

**Files:** Create `src/fxhnt/domain/diversification/hrp.py`; Test `tests/unit/test_hrp.py`

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

Create `tests/unit/test_hrp.py`:
```python
"""HRP allocation: weights sum to 1, all positive, and a duplicated (perfectly-correlated) pair splits the
weight it would get as a single asset (the diversification property)."""
from __future__ import annotations

import numpy as np

from fxhnt.domain.diversification.hrp import hrp_weights


def _series(rng, n, vol):
    return rng.normal(0.0, vol, n)


def test_weights_sum_to_one_all_positive() -> None:
    rng = np.random.default_rng(0)
    rets = {f"S{i}": _series(rng, 500, 0.01) for i in range(5)}
    w = hrp_weights(rets)
    assert abs(sum(w.values()) - 1.0) < 1e-9
    assert all(v >= 0 for v in w.values())
    assert set(w) == set(rets)


def test_lower_vol_gets_more_weight() -> None:
    rng = np.random.default_rng(1)
    rets = {"LOWVOL": _series(rng, 500, 0.005), "HIGHVOL": _series(rng, 500, 0.02)}
    w = hrp_weights(rets)
    assert w["LOWVOL"] > w["HIGHVOL"]             # risk parity: less vol -> more weight


def test_degenerate_single_asset() -> None:
    rng = np.random.default_rng(2)
    w = hrp_weights({"ONLY": _series(rng, 100, 0.01)})
    assert abs(w["ONLY"] - 1.0) < 1e-9


def test_correlated_cluster_shares_weight() -> None:
    # two identical assets + one independent: the identical pair together should not dominate
    rng = np.random.default_rng(3)
    base = _series(rng, 800, 0.01)
    indep = _series(rng, 800, 0.01)
    w = hrp_weights({"A": base, "A_DUP": base.copy(), "B": indep})
    assert abs(sum(w.values()) - 1.0) < 1e-9
    assert w["B"] > w["A"]                        # the independent asset gets more than each twin
```

- [ ] **Step 2: Run (fails)** — `pytest tests/unit/test_hrp.py -v` → ModuleNotFound.

- [ ] **Step 3: Implement**

Create `src/fxhnt/domain/diversification/hrp.py`:
```python
"""Hierarchical Risk Parity (López de Prado 2016). corr → distance → hierarchical linkage →
quasi-diagonalization → recursive bisection with inverse-variance allocation. No matrix inversion, no
concentrated weights. Falls back to inverse-variance on a degenerate/singular correlation."""
from __future__ import annotations

import numpy as np


def _ivp(cov: np.ndarray) -> np.ndarray:
    iv = 1.0 / np.diag(cov)
    return iv / iv.sum()


def _cluster_var(cov: np.ndarray, items: list[int]) -> float:
    sub = cov[np.ix_(items, items)]
    w = _ivp(sub).reshape(-1, 1)
    return float((w.T @ sub @ w)[0, 0])


def _quasi_diag(link: np.ndarray) -> list[int]:
    import pandas as pd  # scipy returns linkage; reproduce LdP's ordering without pandas if unavailable
    link = link.astype(int)
    sort_ix = pd.Series([link[-1, 0], link[-1, 1]])
    num_items = link[-1, 3]
    while sort_ix.max() >= num_items:
        sort_ix.index = range(0, sort_ix.shape[0] * 2, 2)
        df0 = sort_ix[sort_ix >= num_items]
        i, j = df0.index, df0.values - num_items
        sort_ix[i] = link[j, 0]
        df1 = pd.Series(link[j, 1], index=i + 1)
        sort_ix = pd.concat([sort_ix, df1]).sort_index()
        sort_ix.index = range(sort_ix.shape[0])
    return sort_ix.tolist()


def hrp_weights(returns: dict[str, np.ndarray]) -> dict[str, float]:
    syms = list(returns)
    if len(syms) == 1:
        return {syms[0]: 1.0}
    n = min(len(r) for r in returns.values())
    mat = np.array([returns[s][-n:] for s in syms])          # rows = assets
    cov = np.cov(mat)
    if cov.ndim == 0:
        return {syms[0]: 1.0}
    std = np.sqrt(np.diag(cov))
    denom = np.outer(std, std)
    with np.errstate(divide="ignore", invalid="ignore"):
        corr = np.where(denom > 0, cov / denom, 0.0)
    np.fill_diagonal(corr, 1.0)
    try:
        from scipy.cluster.hierarchy import linkage
        from scipy.spatial.distance import squareform
        dist = np.sqrt(np.clip((1.0 - corr) / 2.0, 0.0, 1.0))
        link = linkage(squareform(dist, checks=False), method="single")
        order = _quasi_diag(link)
        w = np.ones(len(syms))
        clusters = [order]
        while clusters:
            clusters = [c[j:k] for c in clusters for j, k in ((0, len(c) // 2), (len(c) // 2, len(c)))
                        if len(c) > 1]
            for i in range(0, len(clusters), 2):
                c0, c1 = clusters[i], clusters[i + 1]
                v0, v1 = _cluster_var(cov, c0), _cluster_var(cov, c1)
                alpha = 1.0 - v0 / (v0 + v1) if (v0 + v1) > 0 else 0.5
                w[c0] *= alpha
                w[c1] *= 1.0 - alpha
        w = w / w.sum()
        return {syms[i]: float(w[i]) for i in range(len(syms))}
    except Exception:                                         # noqa: BLE001 — degenerate -> inverse-variance
        w = _ivp(cov)
        return {syms[i]: float(w[i]) for i in range(len(syms))}
```

Note: if `pandas` is not a dependency, replace `_quasi_diag` with a pure-numpy tree walk. Check `python -c "import pandas"`; if it fails, implement `_quasi_diag` by recursively expanding the scipy linkage tree (the `scipy.cluster.hierarchy.to_tree` API gives a `pre_order()` leaf ordering — prefer that: `from scipy.cluster.hierarchy import to_tree; order = to_tree(link).pre_order()` — this removes the pandas dependency entirely. Use `to_tree(...).pre_order()` and delete `_quasi_diag` + the pandas import.)

- [ ] **Step 4: Run (passes) + commit** — fix the quasi-diag per the note so no pandas dependency; `pytest tests/unit/test_hrp.py -v` → PASS.
```bash
git add src/fxhnt/domain/diversification/hrp.py tests/unit/test_hrp.py
git commit -m "feat(factory): Hierarchical Risk Parity weights (scipy to_tree, ivp fallback)"
```

---

### Task 4: Marginal Sharpe contribution

**Files:** Create `src/fxhnt/domain/diversification/marginal.py`; Test `tests/unit/test_marginal.py`

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

Create `tests/unit/test_marginal.py`:
```python
"""Marginal contribution = the change in book Sharpe from adding a sleeve at a small weight."""
from __future__ import annotations

import numpy as np

from fxhnt.domain.diversification.marginal import marginal_sharpe


def test_uncorrelated_positive_sleeve_adds_sharpe() -> None:
    rng = np.random.default_rng(0)
    book = rng.normal(0.0005, 0.01, 1000)
    sleeve = rng.normal(0.0005, 0.01, 1000)       # independent, positive drift
    assert marginal_sharpe(book, sleeve, weight=0.2) > 0.0


def test_duplicate_book_adds_little() -> None:
    rng = np.random.default_rng(1)
    book = rng.normal(0.0005, 0.01, 1000)
    add = marginal_sharpe(book, book.copy(), weight=0.2)   # same stream -> no diversification
    indep = marginal_sharpe(book, rng.normal(0.0005, 0.01, 1000), weight=0.2)
    assert indep > add


def test_too_short_returns_nan() -> None:
    assert np.isnan(marginal_sharpe(np.array([0.01]), np.array([0.01]), weight=0.2))
```

- [ ] **Step 2: Run (fails)** — `pytest tests/unit/test_marginal.py -v` → ModuleNotFound.

- [ ] **Step 3: Implement**

Create `src/fxhnt/domain/diversification/marginal.py`:
```python
"""Marginal contribution of a candidate sleeve to the book: the increase in annualized Sharpe from blending
it in at `weight`. Positive = the sleeve improves the risk-adjusted book (the selection objective). NaN if
the series are too short to estimate."""
from __future__ import annotations

import numpy as np

_ANN = np.sqrt(252.0)


def _sharpe(x: np.ndarray) -> float:
    x = x[np.isfinite(x)]
    if len(x) < 2 or x.std() == 0:
        return float("nan")
    return float(x.mean() / x.std() * _ANN)


def marginal_sharpe(book: np.ndarray, sleeve: np.ndarray, *, weight: float = 0.2) -> float:
    n = min(len(book), len(sleeve))
    if n < 2:
        return float("nan")
    b, s = book[-n:], sleeve[-n:]
    base = _sharpe(b)
    blended = _sharpe((1.0 - weight) * b + weight * s)
    if not (np.isfinite(base) and np.isfinite(blended)):
        return float("nan")
    return blended - base
```

- [ ] **Step 4: Run (passes) + commit**
```bash
git add src/fxhnt/domain/diversification/marginal.py tests/unit/test_marginal.py
git commit -m "feat(factory): marginal Sharpe contribution (selection objective)"
```

---

### Task 5: Diversification gate

**Files:** Create `src/fxhnt/domain/diversification/gate.py`; Test `tests/unit/test_div_gate.py`

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

Create `tests/unit/test_div_gate.py`:
```python
"""Diversification gate: admit a forward-survivor only if left-tail corr is low AND it adds Sharpe.
Fail-SAFE: if the tail can't be estimated, do NOT admit."""
from __future__ import annotations

import numpy as np

from fxhnt.domain.diversification.gate import diversification_gate


def test_admits_uncorrelated_additive_sleeve() -> None:
    rng = np.random.default_rng(0)
    book = rng.normal(0.0005, 0.01, 1000)
    sleeve = rng.normal(0.0005, 0.01, 1000)
    v = diversification_gate(sleeve, book, max_tail_corr=0.5, min_marginal=0.0, tail_frac=0.1)
    assert v.admit is True


def test_rejects_tail_correlated_sleeve() -> None:
    rng = np.random.default_rng(1)
    book = rng.normal(0.0005, 0.01, 1000)
    sleeve = book.copy()                            # perfectly correlated incl. tail
    v = diversification_gate(sleeve, book, max_tail_corr=0.5, min_marginal=0.0, tail_frac=0.1)
    assert v.admit is False and "tail" in v.reason.lower()


def test_failsafe_when_tail_unestimable() -> None:
    book = np.array([0.01, 0.02, -0.01])           # too few tail days
    sleeve = np.array([0.0, 0.0, 0.0])
    v = diversification_gate(sleeve, book, max_tail_corr=0.5, min_marginal=0.0, tail_frac=0.1)
    assert v.admit is False and "unestimable" in v.reason.lower()
```

- [ ] **Step 2: Run (fails)** — `pytest tests/unit/test_div_gate.py -v` → ModuleNotFound.

- [ ] **Step 3: Implement**

Create `src/fxhnt/domain/diversification/gate.py`:
```python
"""Diversification gate — the selection that makes the factory about UNCORRELATED edges. Admit a
forward-survivor to the deployed book only if (a) its LEFT-TAIL correlation to the book is below the cap and
(b) it adds at least `min_marginal` to book Sharpe. Fail-safe: an unestimable tail correlation is NOT
admitted (don't deploy on unknown crisis-correlation)."""
from __future__ import annotations

from dataclasses import dataclass

import numpy as np

from fxhnt.domain.diversification.correlation import left_tail_correlation
from fxhnt.domain.diversification.marginal import marginal_sharpe


@dataclass(frozen=True, slots=True)
class DiversificationVerdict:
    admit: bool
    left_tail_corr: float
    marginal: float
    reason: str


def diversification_gate(sleeve: np.ndarray, book: np.ndarray, *, max_tail_corr: float,
                         min_marginal: float, tail_frac: float, weight: float = 0.2) -> DiversificationVerdict:
    lt = left_tail_correlation(sleeve, book, tail_frac=tail_frac)
    mg = marginal_sharpe(book, sleeve, weight=weight)
    if not np.isfinite(lt):
        return DiversificationVerdict(False, lt, mg, "left-tail correlation unestimable — fail-safe reject")
    if lt >= max_tail_corr:
        return DiversificationVerdict(False, lt, mg, f"left-tail corr {lt:.2f} ≥ {max_tail_corr}")
    if not np.isfinite(mg) or mg < min_marginal:
        return DiversificationVerdict(False, lt, mg, f"marginal Sharpe {mg:.3f} < {min_marginal}")
    return DiversificationVerdict(True, lt, mg, f"admit: tail-corr {lt:.2f}, +{mg:.3f} Sharpe")
```

- [ ] **Step 4: Run (passes) + commit**
```bash
git add src/fxhnt/domain/diversification/gate.py tests/unit/test_div_gate.py
git commit -m "feat(factory): diversification gate (left-tail corr + marginal, fail-safe)"
```

---

### Task 6: Verify the gauntlet SR* is FST-scaled by the honest global N

**Files:** Test `tests/unit/test_dsr_scales_with_n.py` (verification only; code change ONLY if the test fails)

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

Create `tests/unit/test_dsr_scales_with_n.py`:
```python
"""The multiple-testing bar must RISE with the trial count: the same returns clear DSR at N=1 but not at a
large N (False Strategy Theorem). This locks the immune system the continuous factory depends on."""
from __future__ import annotations

import numpy as np

from fxhnt.domain.gauntlet.core import deflated_sharpe


def test_dsr_drops_as_trials_grow() -> None:
    rng = np.random.default_rng(0)
    r = rng.normal(0.0008, 0.01, 1000)            # a modest positive stream
    sr_var = 0.25                                  # variance of trial Sharpes across the search
    dsr_n1 = deflated_sharpe(r, 1, sr_var)
    dsr_n500 = deflated_sharpe(r, 500, sr_var)
    assert dsr_n500 < dsr_n1                        # more trials -> higher bar -> lower DSR
```

- [ ] **Step 2: Run it** — `pytest tests/unit/test_dsr_scales_with_n.py -v`.
  - If PASS: the gauntlet already implements the FST form. Commit the test (locks the behavior) and move on.
  - If FAIL: open `src/fxhnt/domain/gauntlet/core.py` `deflated_sharpe` and ensure the benchmark `SR*` is the expected-max of N trials, `SR* = sqrt(sr_variance) * ((1-γ)·Z⁻¹(1-1/N) + γ·Z⁻¹(1-1/(N·e)))` (Euler-Mascheroni γ≈0.5772), and the DSR is the PSR evaluated against `SR*`. Implement that, re-run.

- [ ] **Step 3: Commit**
```bash
git add tests/unit/test_dsr_scales_with_n.py src/fxhnt/domain/gauntlet/core.py
git commit -m "test(factory): lock DSR bar rises with trial count (FST immune system)"
```

---

# Milestone 2 — Factory loop + governance

### Task 7: Governance (caps + kill-switch)

**Files:** Create `src/fxhnt/application/factory/__init__.py` (empty), `governance.py`; Test `tests/unit/test_governance.py`

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

Create `tests/unit/test_governance.py`:
```python
"""Governance: how many new sleeves may promote this cycle, given caps + kill-switch."""
from __future__ import annotations

from fxhnt.application.factory.governance import promote_budget


def test_kill_switch_blocks_all() -> None:
    assert promote_budget(kill_switch=True, deployed=3, max_deployed=20, max_new_per_cycle=2) == 0


def test_capacity_and_burst_caps() -> None:
    assert promote_budget(kill_switch=False, deployed=19, max_deployed=20, max_new_per_cycle=2) == 1  # only 1 slot
    assert promote_budget(kill_switch=False, deployed=5, max_deployed=20, max_new_per_cycle=2) == 2    # burst cap
    assert promote_budget(kill_switch=False, deployed=20, max_deployed=20, max_new_per_cycle=2) == 0   # full
```

- [ ] **Step 2: Run (fails)** — `pytest tests/unit/test_governance.py -v`.

- [ ] **Step 3: Implement**

Create `src/fxhnt/application/factory/__init__.py` (empty).

Create `src/fxhnt/application/factory/governance.py`:
```python
"""Promotion governance: bound the autonomous promote-to-capital decision. Kill-switch halts all; otherwise
the budget is min(free deployed slots, per-cycle burst cap)."""
from __future__ import annotations


def promote_budget(*, kill_switch: bool, deployed: int, max_deployed: int, max_new_per_cycle: int) -> int:
    if kill_switch:
        return 0
    free = max(0, max_deployed - deployed)
    return min(free, max(0, max_new_per_cycle))
```

- [ ] **Step 4: Run (passes) + commit**
```bash
git add src/fxhnt/application/factory/__init__.py src/fxhnt/application/factory/governance.py tests/unit/test_governance.py
git commit -m "feat(factory): promotion governance (caps + kill-switch budget)"
```

---

### Task 8: Factory funnel + allocation persistence

**Files:** Modify `src/fxhnt/adapters/persistence/cockpit_models.py`; Create `src/fxhnt/adapters/persistence/factory_store.py`; Test `tests/integration/test_factory_store.py`

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

Create `tests/integration/test_factory_store.py`:
```python
"""Factory store: persist a cycle's funnel counts + the deployed allocation (for the cockpit)."""
from __future__ import annotations

import datetime as dt

from fxhnt.adapters.persistence.factory_store import FactoryStore


def test_funnel_and_allocation_roundtrip() -> None:
    st = FactoryStore("sqlite://")
    at = dt.datetime(2026, 6, 15, 0, 0)
    st.record_cycle(cycle_id="c1", proposed=12, tested=12, survived=3, forward=8, deployed=2,
                    retired=1, global_trials=5000, at=at)
    st.record_allocation("c1", {"A": 0.6, "B": 0.4}, at)
    funnel = st.latest_cycle()
    assert funnel.proposed == 12 and funnel.deployed == 2 and funnel.global_trials == 5000
    alloc = {a.strategy_id: a.weight for a in st.allocation("c1")}
    assert abs(alloc["A"] - 0.6) < 1e-9 and abs(alloc["B"] - 0.4) < 1e-9
```

- [ ] **Step 2: Run (fails)** — `pytest tests/integration/test_factory_store.py -v`.

- [ ] **Step 3: Add the ORM rows**

In `src/fxhnt/adapters/persistence/cockpit_models.py` add (using the existing imports `dt`, `String`, `Integer`, `Float`, `DateTime`, `Mapped`, `mapped_column`, `CockpitBase`):
```python
class FactoryCycleRow(CockpitBase):
    __tablename__ = "factory_cycles"
    cycle_id: Mapped[str] = mapped_column(String(48), primary_key=True)
    proposed: Mapped[int] = mapped_column(Integer)
    tested: Mapped[int] = mapped_column(Integer)
    survived: Mapped[int] = mapped_column(Integer)
    forward: Mapped[int] = mapped_column(Integer)
    deployed: Mapped[int] = mapped_column(Integer)
    retired: Mapped[int] = mapped_column(Integer)
    global_trials: Mapped[int] = mapped_column(Integer)
    at: Mapped[dt.datetime] = mapped_column(DateTime)


class FactoryAllocationRow(CockpitBase):
    __tablename__ = "factory_allocation"
    cycle_id: Mapped[str] = mapped_column(String(48), primary_key=True)
    strategy_id: Mapped[str] = mapped_column(String(64), primary_key=True)
    weight: Mapped[float] = mapped_column(Float)
    at: Mapped[dt.datetime] = mapped_column(DateTime)
```

- [ ] **Step 4: Implement the store**

Create `src/fxhnt/adapters/persistence/factory_store.py`:
```python
"""Persist factory funnel + deployed allocation into the cockpit DB (mission-control)."""
from __future__ import annotations

import datetime as dt

from sqlalchemy import create_engine, desc, select
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool

from fxhnt.adapters.persistence.cockpit_models import CockpitBase, FactoryAllocationRow, FactoryCycleRow


class FactoryStore:
    def __init__(self, dsn: str) -> None:
        kw = {"future": True}
        if dsn == "sqlite://":
            kw |= {"connect_args": {"check_same_thread": False}, "poolclass": StaticPool}
        self._engine = create_engine(dsn, **kw)
        CockpitBase.metadata.create_all(self._engine)

    def record_cycle(self, *, cycle_id, proposed, tested, survived, forward, deployed, retired,
                     global_trials, at) -> None:
        with Session(self._engine) as s:
            s.merge(FactoryCycleRow(cycle_id=cycle_id, proposed=proposed, tested=tested, survived=survived,
                                    forward=forward, deployed=deployed, retired=retired,
                                    global_trials=global_trials, at=at))
            s.commit()

    def record_allocation(self, cycle_id: str, weights: dict[str, float], at: dt.datetime) -> None:
        with Session(self._engine) as s:
            for sid, w in weights.items():
                s.merge(FactoryAllocationRow(cycle_id=cycle_id, strategy_id=sid, weight=w, at=at))
            s.commit()

    def latest_cycle(self) -> FactoryCycleRow | None:
        with Session(self._engine) as s:
            return s.scalars(select(FactoryCycleRow).order_by(desc(FactoryCycleRow.at))).first()

    def allocation(self, cycle_id: str) -> list[FactoryAllocationRow]:
        with Session(self._engine) as s:
            return list(s.scalars(select(FactoryAllocationRow).where(FactoryAllocationRow.cycle_id == cycle_id)))
```

- [ ] **Step 5: Run (passes) + commit**
```bash
git add src/fxhnt/adapters/persistence/cockpit_models.py src/fxhnt/adapters/persistence/factory_store.py tests/integration/test_factory_store.py
git commit -m "feat(factory): funnel + allocation persistence (cockpit DB)"
```

---

### Task 9: Factory loop (diversification-gated promotion + HRP allocation)

**Files:** Create `src/fxhnt/application/factory/factory_loop.py`; Test `tests/integration/test_factory_loop.py`

This is the integration heart. It REUSES `StrategyStore` (status lifecycle, `forward_returns`) and inserts the diversification gate into the FORWARD→DEPLOYED decision, then HRP-allocates the deployed book.

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

Create `tests/integration/test_factory_loop.py`:
```python
"""Factory loop: a forward-survivor uncorrelated to the deployed book promotes (within governance) and the
deployed book is HRP-allocated; a tail-correlated survivor does NOT promote; kill-switch blocks promotion."""
from __future__ import annotations

import numpy as np

from fxhnt.application.factory.factory_loop import FactoryLoop, LoopConfig


class FakeStore:
    """Minimal StrategyStore stand-in keyed by strategy_id with status + forward_returns."""
    def __init__(self, records): self._r = {r["id"]: r for r in records}
    def deployed(self): return [r for r in self._r.values() if r["status"] == "DEPLOYED"]
    def forward(self): return [r for r in self._r.values() if r["status"] == "FORWARD"]
    def set_status(self, sid, status): self._r[sid]["status"] = status
    def returns(self, sid): return np.array(self._r[sid]["returns"])


def _cfg(**kw):
    base = dict(max_tail_corr=0.5, min_marginal=0.0, tail_frac=0.1, min_forward_days=20,
                max_deployed=20, max_new_per_cycle=2, kill_switch=False)
    base.update(kw); return LoopConfig(**base)


def _rec(id, status, returns):
    return {"id": id, "status": status, "returns": returns}


def test_uncorrelated_survivor_promotes_and_allocates() -> None:
    rng = np.random.default_rng(0)
    book = list(rng.normal(0.0005, 0.01, 300))
    cand = list(rng.normal(0.0005, 0.01, 300))                # independent
    store = FakeStore([_rec("DEP", "DEPLOYED", book), _rec("CAND", "FORWARD", cand)])
    loop = FactoryLoop(store, _cfg())
    res = loop.allocate_and_promote()
    assert "CAND" in res.promoted
    assert store._r["CAND"]["status"] == "DEPLOYED"
    assert abs(sum(res.weights.values()) - 1.0) < 1e-9 and set(res.weights) == {"DEP", "CAND"}


def test_tail_correlated_survivor_blocked() -> None:
    rng = np.random.default_rng(1)
    book = list(rng.normal(0.0005, 0.01, 300))
    store = FakeStore([_rec("DEP", "DEPLOYED", book), _rec("CAND", "FORWARD", list(book))])  # identical
    loop = FactoryLoop(store, _cfg())
    res = loop.allocate_and_promote()
    assert "CAND" not in res.promoted and store._r["CAND"]["status"] == "FORWARD"


def test_kill_switch_blocks_promotion() -> None:
    rng = np.random.default_rng(2)
    book = list(rng.normal(0.0005, 0.01, 300)); cand = list(rng.normal(0.0005, 0.01, 300))
    store = FakeStore([_rec("DEP", "DEPLOYED", book), _rec("CAND", "FORWARD", cand)])
    loop = FactoryLoop(store, _cfg(kill_switch=True))
    res = loop.allocate_and_promote()
    assert res.promoted == [] and store._r["CAND"]["status"] == "FORWARD"
```

- [ ] **Step 2: Run (fails)** — `pytest tests/integration/test_factory_loop.py -v`.

- [ ] **Step 3: Implement**

Create `src/fxhnt/application/factory/factory_loop.py`:
```python
"""The factory's promotion + allocation step. Forward-survivors are admitted to the deployed book only if
they pass the diversification gate (left-tail corr + marginal Sharpe) vs the CURRENT deployed book, bounded
by governance; then the whole deployed book is re-allocated by HRP. Pure orchestration over an injected
store (duck-typed: deployed()/forward()/set_status()/returns()) so it tests offline."""
from __future__ import annotations

from dataclasses import dataclass, field

import numpy as np

from fxhnt.application.factory.governance import promote_budget
from fxhnt.domain.diversification.gate import diversification_gate
from fxhnt.domain.diversification.hrp import hrp_weights


@dataclass(frozen=True, slots=True)
class LoopConfig:
    max_tail_corr: float
    min_marginal: float
    tail_frac: float
    min_forward_days: int
    max_deployed: int
    max_new_per_cycle: int
    kill_switch: bool


@dataclass
class LoopResult:
    promoted: list[str] = field(default_factory=list)
    weights: dict[str, float] = field(default_factory=dict)


class FactoryLoop:
    def __init__(self, store, cfg: LoopConfig) -> None:
        self._store = store
        self._cfg = cfg

    def _book_series(self, deployed: list) -> np.ndarray:
        """Equal-weight blend of deployed sleeve returns = the current book series (for the gate)."""
        series = [self._store.returns(r["id"]) for r in deployed]
        if not series:
            return np.array([])
        n = min(len(s) for s in series)
        return np.mean(np.array([s[-n:] for s in series]), axis=0)

    def allocate_and_promote(self) -> LoopResult:
        cfg = self._cfg
        res = LoopResult()
        deployed = self._store.deployed()
        budget = promote_budget(kill_switch=cfg.kill_switch, deployed=len(deployed),
                                max_deployed=cfg.max_deployed, max_new_per_cycle=cfg.max_new_per_cycle)
        book = self._book_series(deployed)
        # rank forward-survivors by marginal contribution; admit the best uncorrelated ones up to budget
        candidates = []
        for r in self._store.forward():
            if len(r["returns"]) < cfg.min_forward_days:
                continue
            sleeve = self._store.returns(r["id"])
            if book.size == 0:                                # empty book: first sleeve admits on data alone
                candidates.append((r["id"], 1.0))
                continue
            v = diversification_gate(sleeve, book, max_tail_corr=cfg.max_tail_corr,
                                     min_marginal=cfg.min_marginal, tail_frac=cfg.tail_frac)
            if v.admit:
                candidates.append((r["id"], v.marginal))
        candidates.sort(key=lambda t: t[1], reverse=True)
        for sid, _ in candidates[:budget]:
            self._store.set_status(sid, "DEPLOYED")
            res.promoted.append(sid)
        # HRP re-allocation over the (updated) deployed book
        deployed_now = self._store.deployed()
        rets = {r["id"]: self._store.returns(r["id"]) for r in deployed_now
                if len(self._store.returns(r["id"])) >= 2}
        res.weights = hrp_weights(rets) if rets else {}
        return res
```

- [ ] **Step 4: Run (passes) + commit**
```bash
git add src/fxhnt/application/factory/factory_loop.py tests/integration/test_factory_loop.py
git commit -m "feat(factory): diversification-gated promotion + HRP allocation loop"
```

---

### Task 10: Wire the loop to the real StrategyStore + CLI

**Files:** Create a thin adapter in `factory_loop.py` (or `factory_store` bridge); Modify `src/fxhnt/cli.py`

- [ ] **Step 1: Add a real-store adapter**

The `FactoryLoop` is duck-typed. Add a small adapter that maps the real `SqlStrategyStore` + the deployed-sleeve return series (from each `StrategyRecord.forward_returns`) to the `deployed()/forward()/set_status()/returns()` interface. Append to `src/fxhnt/application/factory/factory_loop.py`:
```python
class StrategyStoreLoopAdapter:
    """Adapts the real SqlStrategyStore to the FactoryLoop's duck-typed interface."""
    def __init__(self, store) -> None:
        from fxhnt.domain.factory import StrategyStatus
        self._store = store
        self._S = StrategyStatus

    def deployed(self):
        return [{"id": r.strategy_id} for r in self._store.list(status=self._S.DEPLOYED)]

    def forward(self):
        return [{"id": r.strategy_id, "returns": r.forward_returns}
                for r in self._store.list(status=self._S.FORWARD)]

    def set_status(self, sid, status):
        self._store.set_status(sid, self._S(status.lower()) if status.islower() else self._S.DEPLOYED)

    def returns(self, sid):
        import numpy as np
        rec = self._store.get(sid)
        return np.array(rec.forward_returns if rec else [])
```
Note: confirm `StrategyStatus.DEPLOYED.value == "deployed"`; `set_status` here always promotes to DEPLOYED (the only transition the loop makes), so map the literal `"DEPLOYED"` → `StrategyStatus.DEPLOYED` directly. Simplify `set_status` to `self._store.set_status(sid, self._S.DEPLOYED)`.

- [ ] **Step 2: Add CLI commands** (READ the top of `src/fxhnt/cli.py` first for `app`/`get_settings`; adapt). Append:
```python
@app.command("factory-cycle")
def factory_cycle(execute: bool = typer.Option(False, "--execute/--dry-run")) -> None:
    """One factory cycle: diversification-gated promotion + HRP allocation. Dry-run by default."""
    import datetime as dt

    from fxhnt.adapters.persistence import SqlStrategyStore
    from fxhnt.adapters.persistence.factory_store import FactoryStore
    from fxhnt.application.factory.factory_loop import FactoryLoop, LoopConfig, StrategyStoreLoopAdapter
    s = get_settings(); f = s.factory
    store = SqlStrategyStore(s.operational_dsn)
    cfg = LoopConfig(max_tail_corr=f.max_tail_corr, min_marginal=f.min_marginal_sharpe, tail_frac=f.tail_frac,
                     min_forward_days=f.min_forward_days, max_deployed=f.max_deployed_sleeves,
                     max_new_per_cycle=f.max_new_deploys_per_cycle, kill_switch=f.kill_switch or not execute)
    loop = FactoryLoop(StrategyStoreLoopAdapter(store), cfg)
    res = loop.allocate_and_promote()
    fs = FactoryStore(s.operational_dsn)
    cyc = "fc" + dt.datetime.now(dt.UTC).strftime("%Y%m%d")
    if execute:
        fs.record_allocation(cyc, res.weights, dt.datetime.now(dt.UTC).replace(tzinfo=None))
    typer.echo(f"[{cyc}] {'EXECUTED' if execute else 'DRY-RUN'} promoted={res.promoted} "
               f"deployed_weights={ {k: round(v,3) for k,v in res.weights.items()} }")


@app.command("factory-flatten")
def factory_flatten() -> None:
    """Kill-switch: set FXHNT_FACTORY_KILL_SWITCH=true to halt promotions (this just reports current state)."""
    typer.echo(f"factory kill_switch = {get_settings().factory.kill_switch}; "
               f"set FXHNT_FACTORY_KILL_SWITCH=true to halt promotions")
```
(Note: `--dry-run` forces `kill_switch=True` in the cfg so no status changes are written; only `--execute` promotes.)

- [ ] **Step 3: Smoke** — `python3 -c "import fxhnt.cli"` and `fxhnt --help | grep factory-cycle`. Then `pytest -q` (all green) + `ruff check src/fxhnt/application/factory src/fxhnt/domain/diversification` (clean).

- [ ] **Step 4: Commit**
```bash
git add src/fxhnt/application/factory/factory_loop.py src/fxhnt/cli.py
git commit -m "feat(factory): wire loop to SqlStrategyStore + CLI factory-cycle/flatten"
```

---

# Milestone 3 — Domain-analyst fleet

### Task 11: Analyst protocol + fake + fleet

**Files:** Create `src/fxhnt/application/factory/analysts.py`, `analyst_fleet.py`; Test `tests/integration/test_analyst_fleet.py`

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

Create `tests/integration/test_analyst_fleet.py`:
```python
"""Analyst fleet: runs specialists, dedups proposals vs seen, increments the global trial count. A bad
analyst is skipped, not fatal."""
from __future__ import annotations

from fxhnt.application.factory.analyst_fleet import AnalystFleet
from fxhnt.application.factory.analysts import FakeAnalyst, Proposal


class FakeStore:
    def __init__(self): self.seen = set(); self.trials = 0
    def is_seen(self, k): return k in self.seen
    def mark_seen(self, k): self.seen.add(k)
    def add_trials(self, n): self.trials += n


class BoomAnalyst:
    name = "boom"
    def propose(self, universe): raise RuntimeError("boom")


def test_fleet_dedups_and_counts() -> None:
    a = FakeAnalyst("momentum", [Proposal("ES", "trend", {"window": 100.0}),
                                 Proposal("ES", "trend", {"window": 100.0})])  # dup
    store = FakeStore()
    fleet = AnalystFleet([a], store)
    props = fleet.gather(universe=["ES", "NQ"])
    assert len(props) == 1                          # dup removed
    assert store.trials == 1 and len(store.seen) == 1


def test_bad_analyst_is_skipped() -> None:
    good = FakeAnalyst("carry", [Proposal("GC", "carry", {})])
    store = FakeStore()
    props = AnalystFleet([BoomAnalyst(), good], store).gather(universe=["GC"])
    assert len(props) == 1                          # boom skipped, good kept
```

- [ ] **Step 2: Run (fails)** — `pytest tests/integration/test_analyst_fleet.py -v`.

- [ ] **Step 3: Implement**

Create `src/fxhnt/application/factory/analysts.py`:
```python
"""Domain-analyst agents: each specialist proposes hypotheses in its mechanism family. An analyst is any
object with `name: str` and `propose(universe) -> list[Proposal]`. The LLM-backed analysts reason over the
feature warehouse; the FakeAnalyst is for tests (no LLM). All proposals clear the SAME gauntlet downstream —
no special treatment."""
from __future__ import annotations

from dataclasses import dataclass, field


@dataclass(frozen=True, slots=True)
class Proposal:
    market: str
    kind: str
    params: dict[str, float] = field(default_factory=dict)

    def key(self) -> str:
        p = ",".join(f"{k}={v}" for k, v in sorted(self.params.items()))
        return f"{self.market}:{self.kind}:{p}"


class FakeAnalyst:
    """Deterministic analyst for tests — returns a fixed proposal list."""
    def __init__(self, name: str, proposals: list[Proposal]) -> None:
        self.name = name
        self._p = proposals

    def propose(self, universe: list[str]) -> list[Proposal]:
        return list(self._p)


class LlmAnalyst:
    """A domain specialist backed by an LLM. `specialty` shapes the prompt (momentum, carry, ...).
    Reads nothing here beyond the universe + specialty; the warehouse-feature context is injected via
    `context`. Output is parsed into Proposals; malformed lines are dropped (the fleet skips on error)."""
    def __init__(self, specialty: str, chat_model, *, context: str = "") -> None:
        self.name = specialty
        self._chat = chat_model
        self._ctx = context

    def propose(self, universe: list[str]) -> list[Proposal]:
        prompt = (f"You are a {self.name} strategy specialist. Universe: {universe}. {self._ctx}\n"
                  "Propose up to 5 testable (market, kind, params) hypotheses in your specialty, one per line "
                  "as MARKET|KIND|key=val,key=val. Only lines in that exact format.")
        text = self._chat.invoke(prompt).content if hasattr(self._chat.invoke(prompt), "content") else ""
        out: list[Proposal] = []
        for line in str(text).splitlines():
            parts = line.split("|")
            if len(parts) != 3 or parts[0].strip() not in universe:
                continue
            params: dict[str, float] = {}
            for kv in parts[2].split(","):
                if "=" in kv:
                    k, v = kv.split("=", 1)
                    try:
                        params[k.strip()] = float(v)
                    except ValueError:
                        pass
            out.append(Proposal(parts[0].strip(), parts[1].strip(), params))
        return out
```
(Note: simplify the `LlmAnalyst.propose` to call `self._chat.invoke(prompt)` once into a variable, then read `.content`. The double-invoke above is a typo — fix it: `resp = self._chat.invoke(prompt); text = getattr(resp, "content", "")`.)

Create `src/fxhnt/application/factory/analyst_fleet.py`:
```python
"""Runs the analyst fleet: gather proposals across specialists, dedup vs SeenCandidate, count every distinct
hypothesis in the global trial total (the honest N the gauntlet's DSR bar scales with). One bad analyst is
logged and skipped — never fatal to the cycle."""
from __future__ import annotations

import logging

_log = logging.getLogger(__name__)


class AnalystFleet:
    def __init__(self, analysts: list, store) -> None:
        self._analysts = analysts
        self._store = store           # needs is_seen(key)/mark_seen(key)/add_trials(n)

    def gather(self, *, universe: list[str]) -> list:
        fresh = []
        seen_now: set[str] = set()
        for analyst in self._analysts:
            try:
                proposals = analyst.propose(universe)
            except Exception as e:                          # noqa: BLE001 — one bad specialist != no proposals
                _log.warning("analyst %s failed: %s", getattr(analyst, "name", "?"), e)
                continue
            for p in proposals:
                k = p.key()
                if k in seen_now or self._store.is_seen(k):
                    continue
                seen_now.add(k)
                self._store.mark_seen(k)
                fresh.append(p)
        if fresh:
            self._store.add_trials(len(fresh))              # honest global N for the FST bar
        return fresh
```

- [ ] **Step 4: Run (passes) + commit** (fix the `LlmAnalyst.propose` double-invoke noted above before committing; `ruff` must be clean).
```bash
git add src/fxhnt/application/factory/analysts.py src/fxhnt/application/factory/analyst_fleet.py tests/integration/test_analyst_fleet.py
git commit -m "feat(factory): domain-analyst fleet (dedup + honest global trial count; LLM + fake)"
```

---

# Milestone 4 — Cockpit funnel view

### Task 12: Funnel + allocation in the dashboard

**Files:** Modify `src/fxhnt/application/dashboard_service.py`, `src/fxhnt/adapters/web/app.py`, add `src/fxhnt/adapters/web/templates/factory_funnel.html`; Test `tests/integration/test_web_funnel.py`

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

Create `tests/integration/test_web_funnel.py`:
```python
"""The /funnel route renders the latest factory cycle's funnel counts + the deployed HRP weights."""
from __future__ import annotations

import datetime as dt

from fastapi.testclient import TestClient

from fxhnt.adapters.persistence.factory_store import FactoryStore
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.adapters.web.app import create_app


def _client():
    repo = ForwardNavRepo("sqlite://"); repo.migrate()
    fs = FactoryStore("sqlite://")
    at = dt.datetime(2026, 6, 15)
    fs.record_cycle(cycle_id="c1", proposed=20, tested=20, survived=4, forward=9, deployed=3, retired=1,
                    global_trials=12000, at=at)
    fs.record_allocation("c1", {"crypto-mom": 0.5, "carry": 0.3, "vrp": 0.2}, at)
    return TestClient(create_app(repo, factory_store=fs))


def test_funnel_route() -> None:
    r = _client().get("/funnel")
    assert r.status_code == 200
    assert "12000" in r.text and "crypto-mom" in r.text and "deployed" in r.text.lower()
```

- [ ] **Step 2: Run (fails)** — `pytest tests/integration/test_web_funnel.py -v`.

- [ ] **Step 3: Implement** — `create_app` gains an optional `factory_store=None` param; add a `/funnel` route that reads `latest_cycle()` + `allocation(cycle_id)` and renders `factory_funnel.html` (extend `base.html`, show the funnel counts + global trial count + a weights table). Mirror the existing cockpit route/template pattern. `create_app_from_settings` builds a `FactoryStore(get_settings().operational_dsn)` and passes it.

`src/fxhnt/adapters/web/templates/factory_funnel.html`:
```html
{% extends "base.html" %}
{% block title %}factory funnel{% endblock %}
{% block body %}
{% if c %}
<p>cycle <b>{{ c.cycle_id }}</b> · global hypotheses tested: <b>{{ c.global_trials }}</b></p>
<table><thead><tr><th>stage</th><th>count</th></tr></thead><tbody>
  <tr><td>proposed</td><td>{{ c.proposed }}</td></tr>
  <tr><td>tested</td><td>{{ c.tested }}</td></tr>
  <tr><td>gauntlet-survived</td><td>{{ c.survived }}</td></tr>
  <tr><td>forward</td><td>{{ c.forward }}</td></tr>
  <tr><td>deployed</td><td>{{ c.deployed }}</td></tr>
  <tr><td>retired</td><td>{{ c.retired }}</td></tr>
</tbody></table>
<h3>deployed book — HRP weights</h3>
<table><thead><tr><th>sleeve</th><th>weight</th></tr></thead><tbody>
{% for a in alloc %}<tr><td>{{ a.strategy_id }}</td><td>{{ '%.1f'|format(100*a.weight) }}%</td></tr>{% endfor %}
</tbody></table>
{% else %}<p class="muted">no factory cycle recorded yet</p>{% endif %}
{% endblock %}
```

- [ ] **Step 4: Run (passes) + commit**
```bash
git add src/fxhnt/adapters/web/ tests/integration/test_web_funnel.py
git commit -m "feat(factory): cockpit funnel view (funnel counts + global trials + HRP weights)"
```

---

# Milestone 5 — In-cluster scheduling

### Task 13: Factory CronJob

**Files:** Create `infra/k8s/jobs/fxhnt-factory-cronjob.yaml`

- [ ] **Step 1: Write the manifest** — mirror `infra/k8s/jobs/fxhnt-forward-cronjob.yaml` (same image, `imagePullSecrets`, `nodeSelector`, the `FXHNT_OPERATIONAL_DSN`+`PGPASSWORD` env, the `app.kubernetes.io/part-of: foxhunt` + `name: fxhnt-factory` pod labels, the `fxhnt-forward` egress NetworkPolicy pattern). The image must be built `.[web,pg,data,factory]` (add `factory` to the Dockerfile pip install). Schedule `0 1 * * *` (after the 23:30 forward ingest). Command:
```yaml
              command: ["sh", "-c"]
              args:
                - |
                  set -e
                  fxhnt migrate-cockpit
                  fxhnt factory-cycle --execute
```
Add a NetworkPolicy `fxhnt-factory` (egress to postgres:5432 + DNS) mirroring `fxhnt-dashboard` in `infra/k8s/network-policies/fxhnt-cockpit.yaml`, OR reuse the part-of label + add the name to the existing dashboard netpol's selector. Update `infra/docker/fxhnt.Dockerfile` pip install to include `factory`.

- [ ] **Step 2: Validate (dry-run)** — `kubectl apply --dry-run=server -f infra/k8s/jobs/fxhnt-factory-cronjob.yaml` → validates.

- [ ] **Step 3: Commit**
```bash
git add infra/k8s/jobs/fxhnt-factory-cronjob.yaml infra/k8s/network-policies/fxhnt-cockpit.yaml infra/docker/fxhnt.Dockerfile
git commit -m "deploy(factory): daily in-cluster factory-cycle CronJob"
```

> Live deploy (build+push image, kubectl apply, verify) is a held step requiring the registry/cluster, same as the cockpit/exec deploys — run via the Argo build (`./scripts/argo-deploy-cockpit.sh` pattern) on explicit go.

---

## Self-Review notes

- **Spec coverage:** diversification core — correlation/left-tail (T2), HRP (T3), marginal (T4), gate (T5);
  FST immune system (T6); governance (T7); funnel/allocation persistence (T8); diversification-gated
  promotion + HRP allocation loop (T9) + real-store wiring + CLI (T10); analyst fleet (T11); cockpit funnel
  (T12); in-cluster CronJob (T13). All spec sections map to a task.
- **Reuse:** the loop inserts the diversification gate into the EXISTING FORWARD→DEPLOYED transition and
  reuses `StrategyStore` + `ForwardValidationService`'s CUSUM edge-decay (retirement already exists); the
  analyst fleet feeds the EXISTING `DiscoveryService`/gauntlet (which already implements the FST DSR).
- **Type consistency:** `DiversificationVerdict` (T5), `LoopConfig`/`LoopResult` (T9), `Proposal` (T11),
  `FactoryCycleRow`/`FactoryAllocationRow` (T8) used consistently downstream; `hrp_weights`/`diversification_gate`/
  `left_tail_correlation`/`marginal_sharpe`/`promote_budget` signatures match across tasks.
- **No placeholders in code** beyond two explicitly-flagged fix-ups (the `_quasi_diag`→`to_tree().pre_order()`
  pandas-removal in T3, and the `LlmAnalyst.propose` single-invoke fix in T11) — both have the exact correction
  stated inline.
- **Honest scope:** the LLM analysts (T11) reason over injected context; the deep feature-warehouse streaming
  is a documented follow-on (the spec's open follow-ups). v1 produces a working diversification-aware factory
  on the existing hunters + a fleet scaffold ready for richer feature context.
