fix(factory): capital-correct loop — record_cycle, sequential HRP-weighted admission, kill-switch, tests

Fix 1: wire record_cycle in factory-cycle --execute so the funnel table is populated.
  LoopResult gains deployed_before/deployed_after/forward_count; cli.py calls
  fs.record_cycle(proposed=forward_count, ...) and record_allocation after the loop.

Fix 2+4: replace equal-weight book proxy with HRP-weighted blend of deployed sleeves;
  rewrite admission as sequential — each candidate is re-gated against the running mean
  of already-admitted-this-cycle candidates, so two candidates uncorrelated to the book
  but correlated to each other cannot both be admitted in one cycle.

Fix 3: HRP inclusion now requires >= min_forward_days (was >= 2). Folded into the
  final rets comprehension in allocate_and_promote.

Fix 5: StrategyStoreLoopAdapter.set_status now asserts status == "DEPLOYED" so a future
  demotion path cannot silently promote via the loop adapter.

Fix 6: persisted kill-switch — FactoryFlagRow ORM model added to cockpit_models.py;
  FactoryStore.set_kill(on)/is_killed() read/write a key="kill" row; factory-cycle
  folds db kill with env kill; factory-flatten renamed to factory-halt; factory-resume
  and factory-status commands added.

Fix 7: sleeve returns fetched once per sleeve in the final HRP comprehension via a
  shared dict, avoiding double store I/O.

Fix 8: three new integration tests covering min_forward_days filter, empty-book
  founding-admit-then-gate, and mutual-correlation blocking within a cycle.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/fxhnt/adapters/persistence/cockpit_models.py

@@ -80,3 +80,10 @@ class FactoryAllocationRow(CockpitBase):
     strategy_id: Mapped[str] = mapped_column(String(64), primary_key=True)
     weight: Mapped[float] = mapped_column(Float)
     at: Mapped[dt.datetime] = mapped_column(DateTime)
+
+
+class FactoryFlagRow(CockpitBase):
+    """Persisted kill-switch and other factory flags (key/value). key="kill", value="1"/"0"."""
+    __tablename__ = "factory_flags"
+    key: Mapped[str] = mapped_column(String(32), primary_key=True)
+    value: Mapped[str] = mapped_column(String(16))

src/fxhnt/adapters/persistence/factory_store.py

@@ -7,7 +7,7 @@ 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
+from fxhnt.adapters.persistence.cockpit_models import CockpitBase, FactoryAllocationRow, FactoryCycleRow, FactoryFlagRow
 
 
 class FactoryStore:
@@ -39,3 +39,16 @@ class FactoryStore:
     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)))
+
+    # Fix 6: persisted kill-switch
+    def set_kill(self, on: bool) -> None:
+        """Persist the kill flag (key="kill", value="1"/"0") so it survives pod restarts."""
+        with Session(self._engine) as s:
+            s.merge(FactoryFlagRow(key="kill", value="1" if on else "0"))
+            s.commit()
+
+    def is_killed(self) -> bool:
+        """Return True if the persisted kill flag is set. Defaults to False when no row exists."""
+        with Session(self._engine) as s:
+            row = s.scalars(select(FactoryFlagRow).where(FactoryFlagRow.key == "kill")).first()
+            return row is not None and row.value == "1"

src/fxhnt/application/factory/factory_loop.py

@@ -1,7 +1,11 @@
 """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."""
+store (duck-typed: deployed()/forward()/set_status()/returns()) so it tests offline.
+
+Admission is SEQUENTIAL: each candidate is re-checked against the running blend of (HRP-weighted book +
+already-admitted-this-cycle) so two candidates uncorrelated to the book but correlated to each other do not
+both get in."""
 from __future__ import annotations
 
 from dataclasses import dataclass, field
@@ -28,6 +32,9 @@ class LoopConfig:
 class LoopResult:
     promoted: list[str] = field(default_factory=list)
     weights: dict[str, float] = field(default_factory=dict)
+    deployed_before: int = 0
+    deployed_after: int = 0
+    forward_count: int = 0
 
 
 class FactoryLoop:
@@ -35,42 +42,78 @@ class FactoryLoop:
         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:
+    def _weighted_blend(self, rets: dict, weights: dict) -> np.ndarray:
+        """HRP-weighted blend of return series; trims to common tail length."""
+        items = [(rets[k], weights.get(k, 0.0)) for k in rets if len(rets[k]) > 0]
+        if not items:
             return np.array([])
-        n = min(len(s) for s in series)
-        return np.mean(np.array([s[-n:] for s in series]), axis=0)
+        n = min(len(r) for r, _ in items)
+        wsum = sum(w for _, w in items) or 1.0
+        return np.sum([(w / wsum) * r[-n:] for r, w in items], axis=0)
 
     def allocate_and_promote(self) -> LoopResult:
         cfg = self._cfg
         res = LoopResult()
         deployed = self._store.deployed()
+        res.deployed_before = len(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 = []
+
+        # Fix 2+4: start-of-cycle book proxy = HRP-weighted blend of deployed sleeves with enough history
+        dep_rets = {r["id"]: self._store.returns(r["id"]) for r in deployed
+                    if len(self._store.returns(r["id"])) >= cfg.min_forward_days}
+        dep_w = hrp_weights(dep_rets) if dep_rets else {}
+        book = self._weighted_blend(dep_rets, dep_w)
+
+        # Rank eligible survivors by marginal vs the start book
+        survivors = []
         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))
+            if len(sleeve) < cfg.min_forward_days:
+                continue
+            res.forward_count += 1
+            if book.size == 0:
+                # Empty book: founding admit — no correlation to compare against yet
+                survivors.append((r["id"], 1.0, sleeve))
                 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
+                survivors.append((r["id"], v.marginal, sleeve))
+
+        # Sort descending by marginal so the highest-value uncorrelated candidate goes first
+        survivors.sort(key=lambda t: t[1], reverse=True)
+
+        # Fix 2: sequential re-check against the running mean of ALREADY-ADMITTED-THIS-CYCLE candidates.
+        # Comparing only against admitted-this-cycle (not the book) prevents two mutually-correlated
+        # candidates that each pass the initial book gate from both slipping through in one cycle.
+        # The first candidate is admitted freely (book gate was already cleared); subsequent ones must
+        # pass the gate vs the equal-weight mean of previously-admitted sleeves.
+        admitted_this_cycle: list[np.ndarray] = []
+        for sid, _, sleeve in survivors:
+            if len(res.promoted) >= budget:
+                break
+            if not admitted_this_cycle:
+                # First admit of this cycle: book correlation was already cleared above; admit freely.
+                ok = True
+            else:
+                n = min(len(s) for s in admitted_this_cycle)
+                running = np.mean([s[-n:] for s in admitted_this_cycle], axis=0)
+                n2 = min(len(running), len(sleeve))
+                v2 = diversification_gate(sleeve[-n2:], running[-n2:], max_tail_corr=cfg.max_tail_corr,
+                                          min_marginal=cfg.min_marginal, tail_frac=cfg.tail_frac)
+                ok = v2.admit
+            if ok:
+                self._store.set_status(sid, "DEPLOYED")
+                res.promoted.append(sid)
+                admitted_this_cycle.append(sleeve)
+
+        # Fix 3: HRP inclusion requires >= min_forward_days (not >= 2); fetch returns once per sleeve
         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.deployed_after = len(deployed_now)
+        # Fix 7: fetch each sleeve's returns ONCE to avoid double I/O
+        all_rets = {r["id"]: self._store.returns(r["id"]) for r in deployed_now}
+        rets = {sid: arr for sid, arr in all_rets.items() if len(arr) >= cfg.min_forward_days}
         res.weights = hrp_weights(rets) if rets else {}
         return res
 
@@ -89,7 +132,9 @@ class StrategyStoreLoopAdapter:
         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):  # noqa: ARG002 — loop only ever promotes to DEPLOYED
+    def set_status(self, sid, status):
+        # Fix 5: assert the only valid transition so a future demotion path cannot silently promote
+        assert status == "DEPLOYED", f"loop only promotes to DEPLOYED, got {status}"
         self._store.set_status(sid, self._S.DEPLOYED)
 
     def returns(self, sid):

src/fxhnt/cli.py

@@ -485,24 +485,60 @@ def factory_cycle(execute: bool = typer.Option(False, "--execute/--dry-run")) ->
     from fxhnt.application.factory.factory_loop import FactoryLoop, LoopConfig, StrategyStoreLoopAdapter
     s = get_settings(); f = s.factory
     store = SqlStrategyStore(s.operational_dsn)
+    fs = FactoryStore(s.operational_dsn)
+    # Fix 6: effective kill = env flag OR persisted DB flag (OR dry-run mode)
+    effective_kill = f.kill_switch or fs.is_killed() or not execute
     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)
+                     max_new_per_cycle=f.max_new_deploys_per_cycle, kill_switch=effective_kill)
     loop = FactoryLoop(StrategyStoreLoopAdapter(store), cfg)
+    now = dt.datetime.now(dt.UTC).replace(tzinfo=None)
     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} "
+        # Fix 1: populate the funnel table on every execute cycle
+        fs.record_cycle(cycle_id=cyc, proposed=res.forward_count, tested=res.forward_count,
+                        survived=len(res.promoted), forward=res.forward_count, deployed=res.deployed_after,
+                        retired=0, global_trials=store.total_trials(), at=now)
+        fs.record_allocation(cyc, res.weights, now)
+    typer.echo(f"[{cyc}] {'EXECUTED' if execute else 'DRY-RUN'} "
+               f"forward_pool={res.forward_count} promoted={res.promoted} "
+               f"deployed_before={res.deployed_before} deployed_after={res.deployed_after} "
                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")
+@app.command("factory-halt")
+def factory_halt() -> None:
+    """Kill-switch: persist halt flag to DB so all subsequent factory-cycle runs block promotions immediately."""
+    from fxhnt.adapters.persistence.factory_store import FactoryStore
+    fs = FactoryStore(get_settings().operational_dsn)
+    fs.set_kill(True)
+    typer.echo("factory halted (persisted kill=1). Run `fxhnt factory-resume` to re-enable.")
+
+
+@app.command("factory-resume")
+def factory_resume() -> None:
+    """Clear the persisted kill flag so factory-cycle promotions are re-enabled."""
+    from fxhnt.adapters.persistence.factory_store import FactoryStore
+    fs = FactoryStore(get_settings().operational_dsn)
+    fs.set_kill(False)
+    typer.echo("factory resumed (kill=0). Promotions enabled on next factory-cycle --execute.")
+
+
+@app.command("factory-status")
+def factory_status() -> None:
+    """Print the persisted kill state and the latest cycle summary."""
+    from fxhnt.adapters.persistence.factory_store import FactoryStore
+    fs = FactoryStore(get_settings().operational_dsn)
+    killed = fs.is_killed()
+    env_kill = get_settings().factory.kill_switch
+    typer.echo(f"kill_switch: db={killed}  env={env_kill}  effective={killed or env_kill}")
+    cyc = fs.latest_cycle()
+    if cyc:
+        typer.echo(f"latest cycle: {cyc.cycle_id}  proposed={cyc.proposed}  forward={cyc.forward}  "
+                   f"survived={cyc.survived}  deployed={cyc.deployed}  at={cyc.at}")
+    else:
+        typer.echo("latest cycle: (none yet)")
 
 
 if __name__ == "__main__":

tests/integration/test_factory_loop.py

@@ -54,3 +54,60 @@ def test_kill_switch_blocks_promotion() -> None:
     loop = FactoryLoop(store, _cfg(kill_switch=True))
     res = loop.allocate_and_promote()
     assert res.promoted == [] and store._r["CAND"]["status"] == "FORWARD"
+
+
+# Fix 8 — new tests -----------------------------------------------------------
+
+def test_below_min_forward_days_not_promoted() -> None:
+    """A forward survivor with fewer than min_forward_days returns must be silently skipped."""
+    rng = np.random.default_rng(42)
+    book = list(rng.normal(0.0005, 0.01, 300))
+    # Only 10 days of history — below the min_forward_days=20 threshold
+    short = list(rng.normal(0.0005, 0.01, 10))
+    store = FakeStore([_rec("DEP", "DEPLOYED", book), _rec("SHORT", "FORWARD", short)])
+    loop = FactoryLoop(store, _cfg(min_forward_days=20))
+    res = loop.allocate_and_promote()
+    assert "SHORT" not in res.promoted
+    assert store._r["SHORT"]["status"] == "FORWARD"
+    assert res.forward_count == 0  # it never reached the gate
+
+
+def test_empty_book_admits_first_then_gates_rest() -> None:
+    """Empty deployed book: first uncorrelated survivor gets a founding admit, but a duplicate of it
+    must be blocked by the running-blend re-gate on the second candidate."""
+    rng = np.random.default_rng(7)
+    a_returns = list(rng.normal(0.001, 0.01, 100))
+    b_returns = list(a_returns)  # identical to A → corr ≈ 1.0 → must be blocked after A is admitted
+    store = FakeStore([
+        _rec("A", "FORWARD", a_returns),
+        _rec("B", "FORWARD", b_returns),
+    ])
+    loop = FactoryLoop(store, _cfg(max_new_per_cycle=2))
+    res = loop.allocate_and_promote()
+    # Exactly one should be admitted; B is a duplicate of A so it must be blocked
+    assert len(res.promoted) == 1
+    promoted_id = res.promoted[0]
+    other_id = "B" if promoted_id == "A" else "A"
+    assert store._r[promoted_id]["status"] == "DEPLOYED"
+    assert store._r[other_id]["status"] == "FORWARD"
+
+
+def test_mutually_correlated_same_cycle_admits_blocked() -> None:
+    """Two forward survivors uncorrelated to the book but nearly identical to each other:
+    only ONE should be admitted per cycle even though max_new_per_cycle=2."""
+    rng = np.random.default_rng(99)
+    book = list(rng.normal(0.0005, 0.01, 300))
+    # Candidates are independent of the book but almost perfectly correlated with each other
+    base = rng.normal(0.001, 0.01, 300)
+    noise = rng.normal(0, 1e-6, 300)
+    cand_a = list(base)
+    cand_b = list(base + noise)  # corr(cand_a, cand_b) ≈ 1.0
+    store = FakeStore([
+        _rec("DEP", "DEPLOYED", book),
+        _rec("CA",  "FORWARD",  cand_a),
+        _rec("CB",  "FORWARD",  cand_b),
+    ])
+    loop = FactoryLoop(store, _cfg(max_new_per_cycle=2))
+    res = loop.allocate_and_promote()
+    # Both pass the gate vs the book, but the second must fail the running-blend re-gate
+    assert len(res.promoted) == 1

tests/integration/test_factory_store.py

@@ -16,3 +16,13 @@ def test_funnel_and_allocation_roundtrip() -> None:
     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
+
+
+def test_kill_switch_persists() -> None:
+    """set_kill(True) → is_killed() True; set_kill(False) → is_killed() False."""
+    st = FactoryStore("sqlite://")
+    assert st.is_killed() is False  # default: no row → False
+    st.set_kill(True)
+    assert st.is_killed() is True
+    st.set_kill(False)
+    assert st.is_killed() is False